diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 14b0ced207..d59350e064 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -15,6 +15,18 @@ ## Purpose +### Memory annotation is not agent admission + +The shared Memory model's optional guest `observation` records selected metric +state/source/original known time. Missing annotation is legacy/unknown. It +changes neither numeric validity nor enrollment, identity, removal, token scope +or guest-command authority. Canonical agent metric annotations use server-owned +report/source receipt time rather than a client-supplied annotation, and a +PVE-refreshed guest row cannot renew the correlated agent observation. +`TestGuestMemoryObservationWireContract` pins omission, round-trip preservation +and unchanged numeric validity; linked-agent/source-selection controls pin +receipt ownership and offline exclusion. No native liveness/thaw is asserted. + ### TrueNAS probes are not fleet or inventory liveness A separate settings connection probe does not establish runtime provider diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 4dbf5cb604..4a5675e753 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,28 @@ ## Purpose +### Guest memory observations are source-owned + +QEMU and LXC memory carry an optional `observation` with `state` (`current`, +`last-known`, `unavailable`), canonical `source`, and original RFC3339 +`observedAt` when known. Normal TTL cache hits keep the original read time. +Repeated QGA/backup deferrals are last-known, never renewed by a poll or +diagnostic RetrievedAt; expired/missing usage and powered-off guests expose +unavailability without inventing a read time. Legacy unannotated values make +no freshness assertion. Independently selected PVE status/listing or linked +Pulse-agent memory remains distinct from disk or QGA collection state. Agent +receipt time belongs to its source status, not a guest row refreshed by PVE. + +Annotation preserves numeric selection, capacity, retention, guest identity, +agent admission, alert thresholds and existing History recording. It grants +no live-check authority or proof of thaw. Existing legacy state, raw Proxmox +facets and selected canonical memory metrics preserve the discriminator; +selected metric provenance must not be borrowed from another merged facet. +`TestGuestMemoryObservationContract` exercises poll/cache/deferral/expiry/ +resumption through unified and served conversions; wire and source-selection +controls cover omission, cloning and independent live agent data. These are +synthetic source proofs, not native QGA/HAOS acceptance or delivered relief. + ### Unavailable backup-age evaluation is not silent recovery PVE, PBS and mock backup checks share one failure-preserving evaluator. The diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 1395d26b1f..bf835e0494 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -21,6 +21,17 @@ ## Purpose +### Memory provenance does not imply recoverability + +Canonical metric values may carry optional selected guest memory observation +state/source/original known time. The facet follows that numeric source through +cloning and frontend conversion; last-known or unavailable memory is not fresh +backup evidence, a storage-health verdict or confirmation of thaw. This +annotation changes no backup identity, retention, rollup selection, restore +policy or snapshot data. The connected monitoring observation and metric +source-selection controls verify the shared type without weakening recovery +evidence requirements. + ### Latest provider observations do not scan retained evidence payloads The protection-posture reader enumerates provider/scope keys through the existing diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index b6ad600f88..4ad877c1d1 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -23,6 +23,23 @@ and sort the complete canonical change table while startup and ingestion wait. ## Purpose +### Guest memory provenance follows metric selection + +Raw Proxmox `memory.observation` preserves the guest builder's state, source +and original known observation time. Canonical memory metrics carry their own +optional `observation` through clones, merges and served frontend conversion. +A retained platform facet cannot supply provenance for a different selected +agent metric. Agent memory annotations are authored from server-owned source +receipt/freshness, never agent-supplied annotation or merged guest LastSeen. +Linked-agent views retain that source time after host/guest correlation. +Absent annotations are unknown/legacy; absent observation times stay absent. + +Numeric preference, unavailable same-source withdrawal, cross-source fallback, +identity and freshness policy remain unchanged. Guest memory and disk-read +provenance are independent. Adapter/source-selection and linked-view controls +exercise these boundaries, with the connected monitoring contract pinning the +actual served projection; this does not prove native recovery or publication. + ### PBS capacity consumers preserve explicit read failure `ResourcePBSDatastore.error` describes existing provider read-failure evidence, diff --git a/internal/models/converters.go b/internal/models/converters.go index 4b272130ed..a6a8242ae2 100644 --- a/internal/models/converters.go +++ b/internal/models/converters.go @@ -1217,10 +1217,11 @@ type ResourceConvertInput struct { // ResourceMetricInput represents a metric value for resource conversion. type ResourceMetricInput struct { - Current float64 - Total *int64 - Used *int64 - Free *int64 + Current float64 + Total *int64 + Used *int64 + Free *int64 + Observation MemoryObservation } type ResourceAlertInput struct { @@ -1321,10 +1322,11 @@ func ConvertResourceToFrontend(input ResourceConvertInput) ResourceFrontend { if input.Memory != nil { rf.Memory = &ResourceMetricFrontend{ - Current: input.Memory.Current, - Total: input.Memory.Total, - Used: input.Memory.Used, - Free: input.Memory.Free, + Current: input.Memory.Current, + Total: input.Memory.Total, + Used: input.Memory.Used, + Free: input.Memory.Free, + Observation: input.Memory.Observation, } } diff --git a/internal/models/metrics_types_test.go b/internal/models/metrics_types_test.go index 64b5755a09..86d4df09d0 100644 --- a/internal/models/metrics_types_test.go +++ b/internal/models/metrics_types_test.go @@ -7,6 +7,43 @@ import ( "time" ) +func TestGuestMemoryObservationWireContract(t *testing.T) { + legacy := Memory{Total: 100, Used: 25, Free: 75, Usage: 25} + payload, err := json.Marshal(legacy) + if err != nil || strings.Contains(string(payload), "observation") { + t.Fatalf("legacy memory unexpectedly asserts freshness: %s / %v", payload, err) + } + at := time.Date(2026, time.October, 4, 12, 0, 0, 123, time.UTC) + for _, state := range []string{"current", "last-known", "unavailable"} { + t.Run(state, func(t *testing.T) { + memory := legacy + memory.Observation = MemoryObservation{State: state, Source: "guest-agent-meminfo", ObservedAt: at} + if state == "unavailable" { + memory.Observation.ObservedAt = time.Time{} + } + payload, err := json.Marshal(memory) + if err != nil { + t.Fatal(err) + } + var roundtrip Memory + if err := json.Unmarshal(payload, &roundtrip); err != nil || roundtrip != memory { + t.Fatalf("memory provenance changed in wire roundtrip: %s / %v", payload, err) + } + if strings.Contains(string(payload), "observedAt") != (state != "unavailable") { + t.Fatalf("unknown time was manufactured: %s", payload) + } + if memory.HasKnownUsage() != legacy.HasKnownUsage() { + t.Fatal("annotation changed numeric validity / threshold policy") + } + metric := ResourceMetricInput{Current: 25, Observation: memory.Observation} + front := ConvertResourceToFrontend(ResourceConvertInput{Memory: &metric, CPU: &metric, Disk: &metric}) + if front.Memory == nil || front.Memory.Observation != memory.Observation || front.CPU.Observation.State != "" || front.Disk.Observation.State != "" { + t.Fatal("memory projection lost evidence or attributed it to another metric") + } + }) + } +} + func TestMetricPoint_ZeroValue(t *testing.T) { var point MetricPoint diff --git a/internal/models/models.go b/internal/models/models.go index 08530261fb..aff44ce977 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -3085,6 +3085,17 @@ type PMGQueueStatus struct { UpdatedAt time.Time `json:"updatedAt"` // When this queue data was collected } +// MemoryObservation describes the provenance of a selected guest reading, not +// guest-agent liveness or permission to run a diagnostic. Current may include a +// normal TTL cache hit; ObservedAt remains the original source observation. +// Last-known values are retained evidence, not a new measurement. A missing +// observation is legacy/unknown, never an assertion of current freshness. +type MemoryObservation struct { + State string `json:"state"` // current, last-known, unavailable + Source string `json:"source"` + ObservedAt time.Time `json:"observedAt,omitzero"` +} + // Memory represents memory usage type Memory struct { Total int64 `json:"total"` @@ -3095,11 +3106,12 @@ type Memory struct { // UsageUnavailable is true when total capacity is known but no cache-aware // source can establish current usage. Consumers must not interpret the // zero-valued Used, Free, or Usage fields as a measurement in that state. - UsageUnavailable bool `json:"usageUnavailable,omitempty"` - Usage float64 `json:"usage"` - Balloon int64 `json:"balloon,omitempty"` - SwapUsed int64 `json:"swapUsed,omitempty"` - SwapTotal int64 `json:"swapTotal,omitempty"` + UsageUnavailable bool `json:"usageUnavailable,omitempty"` + Usage float64 `json:"usage"` + Balloon int64 `json:"balloon,omitempty"` + SwapUsed int64 `json:"swapUsed,omitempty"` + SwapTotal int64 `json:"swapTotal,omitempty"` + Observation MemoryObservation `json:"observation,omitzero"` } // HasKnownUsage reports whether this memory value is safe for live diff --git a/internal/models/models_frontend.go b/internal/models/models_frontend.go index 443c15301a..08438e51cf 100644 --- a/internal/models/models_frontend.go +++ b/internal/models/models_frontend.go @@ -1140,10 +1140,11 @@ func (r ResourceFrontend) NormalizeCollections() ResourceFrontend { // ResourceMetricFrontend represents a metric value for the frontend. type ResourceMetricFrontend struct { - Current float64 `json:"current"` - Total *int64 `json:"total,omitempty"` - Used *int64 `json:"used,omitempty"` - Free *int64 `json:"free,omitempty"` + Current float64 `json:"current"` + Total *int64 `json:"total,omitempty"` + Used *int64 `json:"used,omitempty"` + Free *int64 `json:"free,omitempty"` + Observation MemoryObservation `json:"observation,omitzero"` } // ResourceNetworkFrontend represents network metrics for the frontend. diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 30eeb870b5..165f94a540 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2980,6 +2980,12 @@ func TestGuestAgentBackupMonitoringContract(t *testing.T) { testGuestAgentBackupMonitoringLifecycle(t) } +func TestGuestMemoryObservationContract(t *testing.T) { + t.Run("poll-to-served-observation", testGuestMemoryObservationLifecycle) + t.Run("identity-and-origin-boundaries", testGuestMemoryObservationKeepsOriginsSeparate) + t.Run("independent-linked-agent", testGuestMemoryObservationPreservesIndependentLinkedAgent) +} + // Same-target, unambiguous lock evidence is required even with warm caches. func TestGuestAgentBackupLockEvidenceContract(t *testing.T) { for _, test := range []struct{ name, config string }{ diff --git a/internal/monitoring/guest_agent_backup_integration_test.go b/internal/monitoring/guest_agent_backup_integration_test.go index d9e7dcf9bc..b40699cded 100644 --- a/internal/monitoring/guest_agent_backup_integration_test.go +++ b/internal/monitoring/guest_agent_backup_integration_test.go @@ -155,7 +155,7 @@ func testGuestAgentBackupMonitoringLifecycleWithUnverifiedConfig(t *testing.T, u if locked.ID != initial.ID || locked.GuestAgentStatus != "deferred" || locked.DiskStatusReason != "prev-vm-locked" { t.Fatalf("locked poll %d lost continuity/truth: %#v", poll, locked) } - if locked.Memory != initial.Memory || source != "previous-snapshot" || locked.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(locked.NetworkInterfaces, initial.NetworkInterfaces) { + if guestMemoryValuesOnly(locked.Memory) != guestMemoryValuesOnly(initial.Memory) || source != "previous-snapshot" || locked.Disk.Used != initial.Disk.Used || !reflect.DeepEqual(locked.NetworkInterfaces, initial.NetworkInterfaces) { t.Fatalf("locked poll %d lost last-known evidence: source=%s memory=%#v want=%#v", poll, source, locked.Memory, initial.Memory) } } @@ -176,7 +176,7 @@ func testGuestAgentBackupMonitoringLifecycleWithUnverifiedConfig(t *testing.T, u t.Fatal(err) } locked, _ = build() - if calls.Load() != before || locked.GuestAgentStatus != "deferred" || locked.AgentVersion != initial.AgentVersion || locked.Memory != initial.Memory { + if calls.Load() != before || locked.GuestAgentStatus != "deferred" || locked.AgentVersion != initial.AgentVersion || guestMemoryValuesOnly(locked.Memory) != guestMemoryValuesOnly(initial.Memory) { t.Fatalf("resource-lock fallback queried/lost identity: calls=%d %#v", calls.Load(), locked) } // The per-node inventory fallback must propagate its lock to the same builder. @@ -240,3 +240,9 @@ func TestGuestAgentDeferralPreservesMetadataAndMemoryCache(t *testing.T) { t.Fatal("deferral queued work, renewed cache or poisoned supported-OS evidence") } } + +// Provenance is asserted separately; numeric continuity includes every old field. +func guestMemoryValuesOnly(memory models.Memory) models.Memory { + memory.Observation = models.MemoryObservation{} + return memory +} diff --git a/internal/monitoring/guest_memory_observation.go b/internal/monitoring/guest_memory_observation.go new file mode 100644 index 0000000000..aa29e34b78 --- /dev/null +++ b/internal/monitoring/guest_memory_observation.go @@ -0,0 +1,97 @@ +package monitoring + +import ( + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +type guestMemoryObservationContext struct { + source string + providerObservedAt time.Time + deferred bool + previous *GuestMemorySnapshot + linkedAgent models.Host +} + +// guestMemoryObservation annotates the value already selected by the builder. +// It does not select different numeric evidence, change alert/history policy, +// renew a cache, or run a guest command. In particular disk state is not memory +// provenance: PVE or a linked Pulse agent may still supply an independent read. +func (m *Monitor) guestMemoryObservation(instance, guestType, node string, vmid int, memory models.Memory, evidence guestMemoryObservationContext, now time.Time) models.MemoryObservation { + source := CanonicalMemorySource(evidence.source) + if !memory.HasKnownUsage() || source == "unavailable" || source == "powered-off" { + if source != "powered-off" { + source = "unavailable" + } + return models.MemoryObservation{State: "unavailable", Source: source} + } + observation := models.MemoryObservation{State: "current", Source: source} + switch source { + case "guest-agent-meminfo", "guest-agent-meminfo-derived", "previous-snapshot": + // The cache, rather than the repeatedly refreshed diagnostic/poll + // timestamp, owns the age of retained QGA evidence. + previous := evidence.previous + sameGuest := previous != nil && previous.Instance == instance && previous.GuestType == guestType && previous.Node == node && previous.VMID == vmid + qgaOrigin := source != "previous-snapshot" || (sameGuest && + (previous.Memory.Observation.Source == "guest-agent-meminfo" || previous.Memory.Observation.Source == "guest-agent-meminfo-derived" || + (previous.Memory.Observation.State == "" && (CanonicalMemorySource(previous.MemorySource) == "guest-agent-meminfo" || CanonicalMemorySource(previous.MemorySource) == "guest-agent-meminfo-derived")))) + if guestType == "qemu" && qgaOrigin { + if original, ok := m.cachedGuestMemoryObservation(instance, node, vmid, memory, now); ok { + observation = original + } + } + if source == "previous-snapshot" { + observation.State = "last-known" + if observation.ObservedAt.IsZero() && sameGuest && previous.Memory.Total == memory.Total && previous.Memory.Used == memory.Used { + // An annotated prior source carries its original observation. + // A legacy previous-snapshot's RetrievedAt cannot supply it. + if previous.Memory.Observation.State == "current" || previous.Memory.Observation.State == "last-known" { + observation = previous.Memory.Observation + observation.State = "last-known" + } + } + } else if evidence.deferred || (!observation.ObservedAt.IsZero() && now.Sub(observation.ObservedAt) >= vmAgentMemCacheTTL) { + observation.State = "last-known" + } + case "agent": + observation.ObservedAt = evidence.linkedAgent.LastSeen + if original := evidence.linkedAgent.Memory.Observation; original.State != "" { + observation = original + } + case "available-field", "derived-free-buffers-cached", "derived-total-minus-used", "status-mem", "status-freemem", "cluster-resources": + observation.ObservedAt = evidence.providerObservedAt + default: + // Unknown provenance cannot be upgraded to a current observation by + // a new poll. Keep the selected numbers and expose the uncertainty. + observation.State = "last-known" + } + if observation.ObservedAt.IsZero() || observation.ObservedAt.After(now) { + observation.ObservedAt = time.Time{} + observation.State = "last-known" + } + return observation +} + +func (m *Monitor) cachedGuestMemoryObservation(instance, node string, vmid int, memory models.Memory, now time.Time) (models.MemoryObservation, bool) { + m.rrdCacheMu.RLock() + entry, ok := m.vmAgentMemCache[guestMemoryCacheKey(instance, node, vmid)] + m.rrdCacheMu.RUnlock() + if !ok || entry.negative || entry.fetchedAt.IsZero() || entry.fetchedAt.After(now) || now.Sub(entry.fetchedAt) > vmAgentMemCleanupMaxAge || memory.Total <= 0 || entry.info.EffectiveAvailable > uint64(memory.Total) || memory.Used != memory.Total-int64(entry.info.EffectiveAvailable) { + return models.MemoryObservation{}, false + } + expected := models.Memory{Free: int64(entry.info.EffectiveAvailable)} + splitReclaimableMemory(&expected, entry.info.Free) + if memory.Free != expected.Free || memory.Cache != expected.Cache { + return models.MemoryObservation{}, false + } + source := "guest-agent-meminfo" + switch entry.info.Source { + case "meminfo-available": + case "meminfo-derived": + source = "guest-agent-meminfo-derived" + default: + return models.MemoryObservation{}, false + } + return models.MemoryObservation{State: "current", Source: source, ObservedAt: entry.fetchedAt}, true +} diff --git a/internal/monitoring/guest_memory_observation_test.go b/internal/monitoring/guest_memory_observation_test.go new file mode 100644 index 0000000000..41fb665b42 --- /dev/null +++ b/internal/monitoring/guest_memory_observation_test.go @@ -0,0 +1,237 @@ +package monitoring + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +// Use the served JSON rather than the proposed Go type so this control also +// compiles against the parent that loses observation provenance entirely. +func assertGuestMemoryObservation(t *testing.T, memory models.Memory, state, source string, at time.Time) { + t.Helper() + data, err := json.Marshal(memory) + if err != nil { + t.Fatal(err) + } + var wire struct { + Observation struct { + State string `json:"state"` + Source string `json:"source"` + ObservedAt *time.Time `json:"observedAt"` + } `json:"observation"` + } + if err := json.Unmarshal(data, &wire); err != nil { + t.Fatal(err) + } + got := wire.Observation + if got.State != state || got.Source != source { + t.Errorf("memory observation = %s, want %s / %s", data, state, source) + } + if at.IsZero() { + if got.ObservedAt != nil { + t.Errorf("unobserved memory acquired a timestamp: %s", data) + } + } else if got.ObservedAt == nil || !got.ObservedAt.Equal(at) { + t.Errorf("memory observation time = %v, want original %v", got.ObservedAt, at) + } +} + +func testGuestMemoryObservationLifecycle(t *testing.T) { + const gib = uint64(1024 * 1024 * 1024) + identity := makeGuestID("provenance", "node", 105) + registry := unifiedresources.NewRegistry(nil) + store := unifiedresources.NewMonitorAdapter(registry) + m := &Monitor{guestMetadataLimiter: make(map[string]time.Time), rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), resourceStore: store} + client := &vmMemoryTrustStubClient{stubPVEClient: &stubPVEClient{}, vmAgentMemAvailable: 5 * gib, + vmStatus: &proxmox.VMStatus{Status: "running", MaxMem: 8 * gib, Mem: 8 * gib, Agent: proxmox.VMAgentField{Value: 1}}} + res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 8 * gib, Mem: 8 * gib} + var previous *models.VM + build := func() models.VM { + t.Helper() + vm, raw, source, notes, at, ok := m.buildVMFromClusterResource(context.Background(), "provenance", res, client, identity, nil, previous) + if !ok { + t.Fatal("guest disappeared") + } + m.recordGuestSnapshot("provenance", "qemu", "node", 105, GuestMemorySnapshot{Status: vm.Status, RetrievedAt: at, MemorySource: source, Memory: vm.Memory, Raw: raw, Notes: notes}) + m.recordGuestMetrics([]models.VM{vm}, nil, time.Now().Add(-time.Second)) + store.PopulateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + views := store.VMs() + if len(views) != 1 { + t.Fatal("guest did not reach unified view") + } + next := previousVMFromView(views[0]) + previous = &next + return vm + } + assertProjection := func(vm models.VM, state, source string, at time.Time) { + t.Helper() + assertGuestMemoryObservation(t, vm.Memory, state, source, at) + legacy := vm.ToFrontend() + if legacy.Memory == nil { + t.Fatal("legacy projection lost memory") + } + assertGuestMemoryObservation(t, *legacy.Memory, state, source, at) + // REST and WebSocket state use this production conversion, including + // registry cloning and the nested Proxmox payload the drawer consumes. + front := m.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + data, err := json.Marshal(front.Resources) + if err != nil { + t.Fatal(err) + } + var wire []struct { + Memory *models.ResourceMetricFrontend `json:"memory"` + Proxmox struct { + Memory models.Memory `json:"memory"` + } `json:"proxmox"` + } + if err := json.Unmarshal(data, &wire); err != nil || len(wire) != 1 { + t.Fatalf("broadcast guest memory unavailable: %s / %v", data, err) + } + assertGuestMemoryObservation(t, wire[0].Proxmox.Memory, state, source, at) + if vm.Memory.HasKnownUsage() { + if wire[0].Memory == nil { + t.Fatal("selected memory metric disappeared") + } + assertGuestMemoryObservation(t, models.Memory{Observation: wire[0].Memory.Observation}, state, source, at) + } else if wire[0].Memory != nil { + t.Fatal("unavailable memory became a numeric metric") + } + if wire[0].Proxmox.Memory.Used != vm.Memory.Used || wire[0].Proxmox.Memory.UsageUnavailable != vm.Memory.UsageUnavailable { + t.Fatal("provenance changed numeric/unknown memory projection") + } + } + initial := build() + key := guestMemoryCacheKey("provenance", "node", 105) + original := m.vmAgentMemCache[key].fetchedAt + if original.IsZero() || initial.Memory.Used != int64(3*gib) { + t.Fatal("successful QGA observation missing") + } + assertProjection(initial, "current", "guest-agent-meminfo", original) + cached := build() + assertProjection(cached, "current", "guest-agent-meminfo", original) + if client.vmAgentMemCalls != 1 { + t.Fatal("normal cache caused a new guest read") + } + res.Lock, client.vmStatus.Lock = "backup", "backup" + for poll := 0; poll < 4; poll++ { + locked := build() + assertProjection(locked, "last-known", "guest-agent-meminfo", original) + if locked.Memory.Used != initial.Memory.Used || m.vmAgentMemCache[key].fetchedAt != original || client.vmAgentMemCalls != 1 { + t.Fatal("deferral changed evidence, its age, or guest read admission") + } + } + if got := len(m.metricsHistory.GetGuestMetrics(identity, "memory", time.Hour)); got != 2 { + t.Errorf("deferred memory became fresh History points: %d, want two normal polls", got) + } + entry := m.vmAgentMemCache[key] + entry.fetchedAt = time.Now().Add(-vmAgentMemCleanupMaxAge - time.Second) + m.vmAgentMemCache[key] = entry + expired := build() + assertProjection(expired, "unavailable", "unavailable", time.Time{}) + if expired.Memory.HasKnownUsage() || client.vmAgentMemCalls != 1 { + t.Fatal("expired evidence became usage or queued a guest read") + } + res.Lock, client.vmStatus.Lock = "", "" + client.vmAgentMemAvailable = 4 * gib + resumed := build() + resumedAt := m.vmAgentMemCache[key].fetchedAt + assertProjection(resumed, "current", "guest-agent-meminfo", resumedAt) + if !resumedAt.After(original) || resumed.Memory.Used != int64(4*gib) || client.vmAgentMemCalls != 2 { + t.Fatal("resumption did not replace original guest observation") + } + // Backup/disk deferral must not disqualify an independently current PVE + // memory reading. Its timestamp belongs to that source, not this poll. + statusAt := time.Now().Add(-time.Second) + res.Lock, client.vmStatus.Lock = "backup", "backup" + client.vmStatus.ObservedAt = statusAt + client.vmStatus.MemInfo = &proxmox.VMMemInfo{Total: 8 * gib, Available: 6 * gib} + live := build() + assertProjection(live, "current", "available-field", statusAt) + if live.Memory.Used != int64(2*gib) || live.GuestAgentStatus != "deferred" || client.vmAgentMemCalls != 2 { + t.Fatal("independent PVE memory was replaced by deferred QGA evidence") + } + res.Status = "stopped" + stopped := build() + assertProjection(stopped, "unavailable", "powered-off", time.Time{}) + if stopped.Memory.Used != 0 { + t.Fatal("powered-off numeric compatibility changed") + } +} + +func testGuestMemoryObservationKeepsOriginsSeparate(t *testing.T) { + now := time.Now() + at := now.Add(-2 * time.Minute) + observed := models.Memory{Total: 100, Used: 25, Free: 75, Usage: 25} + previous := GuestMemorySnapshot{Instance: "site", GuestType: "qemu", Node: "node", VMID: 105, Status: "running", RetrievedAt: now, + MemorySource: "previous-snapshot", Memory: observed} + previous.Memory.Observation = models.MemoryObservation{State: "last-known", Source: "available-field", ObservedAt: at} + m := &Monitor{vmAgentMemCache: map[string]agentMemCacheEntry{guestMemoryCacheKey("site", "node", 105): { + fetchedAt: now.Add(-time.Second), info: proxmox.LinuxMemoryAvailability{Source: "meminfo-available", EffectiveAvailable: 75}}}} + for _, tc := range []struct { + name string + change func(*GuestMemorySnapshot) + source string + at time.Time + }{ + {name: "identical numeric QGA cache cannot replace PVE provenance", source: "available-field", at: at}, + {name: "poll timestamp does not renew legacy retained evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.Memory.Observation = models.MemoryObservation{} }}, + {name: "foreign guest snapshot cannot lend evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.VMID++ }}, + {name: "foreign site snapshot cannot lend evidence", source: "previous-snapshot", change: func(p *GuestMemorySnapshot) { p.Instance = "other" }}, + {name: "future original time remains unknown", source: "available-field", change: func(p *GuestMemorySnapshot) { p.Memory.Observation.ObservedAt = now.Add(time.Second) }}, + } { + t.Run(tc.name, func(t *testing.T) { + prior := previous + if tc.change != nil { + tc.change(&prior) + } + memory := observed + memory.Observation = m.guestMemoryObservation("site", "qemu", "node", 105, memory, guestMemoryObservationContext{source: "previous-snapshot", previous: &prior}, now) + assertGuestMemoryObservation(t, memory, "last-known", tc.source, tc.at) + }) + } +} + +func testGuestMemoryObservationPreservesIndependentLinkedAgent(t *testing.T) { + at := time.Now().Add(-30 * time.Second) + resource := &unifiedresources.Resource{Type: unifiedresources.ResourceTypeVM, LastSeen: time.Now(), + Agent: &unifiedresources.AgentData{Memory: &unifiedresources.AgentMemoryMeta{Total: 100, Used: 40, Free: 60}}, + SourceStatus: map[unifiedresources.DataSource]unifiedresources.SourceStatus{unifiedresources.SourceAgent: {Status: "online", LastSeen: at}}} + view := unifiedresources.NewVMView(resource) + agentMemory, ok := view.LinkedAgentMemory() + if !ok { + t.Fatal("source-owned linked memory missing") + } + for _, guestType := range []string{"qemu", "lxc"} { + t.Run(guestType, func(t *testing.T) { + m := &Monitor{rateTracker: NewRateTracker(), guestMetadataLimiter: make(map[string]time.Time)} + client := &vmMemoryTrustStubClient{stubPVEClient: &stubPVEClient{}, vmStatus: &proxmox.VMStatus{Status: "running", Lock: "backup", MaxMem: 100, Mem: 100}} + res := proxmox.ClusterResource{Type: guestType, Node: "node", VMID: 105, Status: "running", Lock: "backup", MaxMem: 100, Mem: 100, ObservedAt: time.Now()} + identity := makeGuestID("site", "node", 105) + linked := map[string]models.Host{identity: {Memory: agentMemory, Status: "online"}} + var memory models.Memory + if guestType == "qemu" { + vm, _, source, _, _, ok := m.buildVMFromClusterResource(context.Background(), "site", res, client, identity, linked, nil) + if !ok || source != "agent" { + t.Fatalf("linked VM agent was not selected: %s", source) + } + memory = vm.Memory + } else { + ct, _, source, _, ok := m.buildContainerFromClusterResource(context.Background(), "site", res, client, nil, linked) + if !ok || source != "agent" { + t.Fatalf("linked LXC agent was not selected: %s", source) + } + memory = ct.Memory + } + assertGuestMemoryObservation(t, memory, "current", "agent", at) + if memory.Used != 40 || client.vmAgentMemCalls != 0 { + t.Fatal("independent memory changed or caused a guest-agent read") + } + }) + } +} diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 7a1f8b2405..6e9091900e 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -6985,7 +6985,7 @@ func monitorMetricInput(metric *unifiedresources.MetricValue) *models.ResourceMe current = math.Max(metric.Percent, metric.Value) } - result := &models.ResourceMetricInput{Current: current} + result := &models.ResourceMetricInput{Current: current, Observation: metric.Observation} if metric.Total != nil { total := *metric.Total result.Total = &total diff --git a/internal/monitoring/monitor_pve_guest_builders.go b/internal/monitoring/monitor_pve_guest_builders.go index e5cb6e215f..c07138148a 100644 --- a/internal/monitoring/monitor_pve_guest_builders.go +++ b/internal/monitoring/monitor_pve_guest_builders.go @@ -468,6 +468,15 @@ func (m *Monitor) buildVMFromClusterResource( snapshotNotes = append(snapshotNotes, "preserved-original-guest-memory-during-deferral") } } + providerObservedAt := observedAtOr(res.ObservedAt, sampleTime) + if state.detailedStatus != nil && CanonicalMemorySource(state.memorySource) != "cluster-resources" { + providerObservedAt = observedAtOr(state.detailedStatus.ObservedAt, sampleTime) + } + memory.Observation = m.guestMemoryObservation(instanceName, "qemu", res.Node, res.VMID, memory, guestMemoryObservationContext{ + source: state.memorySource, providerObservedAt: providerObservedAt, + deferred: state.guestAgentStatus == "deferred", previous: prevSnapshot, + linkedAgent: vmIDToHostAgent[guestID], + }, sampleTime) vm := models.VM{ ID: guestID, diff --git a/internal/monitoring/monitor_pve_guest_lxc.go b/internal/monitoring/monitor_pve_guest_lxc.go index 6654435e9c..83fa5a44bf 100644 --- a/internal/monitoring/monitor_pve_guest_lxc.go +++ b/internal/monitoring/monitor_pve_guest_lxc.go @@ -99,8 +99,9 @@ func (m *Monitor) buildContainerFromClusterResource( agentHost, hasAgent, ) + previousMemory := m.previousGuestSnapshot(instanceName, "lxc", res.Node, res.VMID) memUsed, memorySource, _ = stabilizeGuestLowTrustMemory( - m.previousGuestSnapshot(instanceName, "lxc", res.Node, res.VMID), + previousMemory, res.Status, memorySource, memTotal, @@ -129,6 +130,10 @@ func (m *Monitor) buildContainerFromClusterResource( Usage: safePercentage(float64(clampedMemUsed), float64(memTotal)), } } + memory.Observation = m.guestMemoryObservation(instanceName, "lxc", res.Node, res.VMID, memory, guestMemoryObservationContext{ + source: memorySource, providerObservedAt: observedAtOr(res.ObservedAt, sampleTime), + previous: previousMemory, linkedAgent: agentHost, + }, time.Now()) diskUsed := res.Disk if diskUsed > res.MaxDisk && res.MaxDisk > 0 { diskUsed = res.MaxDisk diff --git a/internal/unifiedresources/clone_test.go b/internal/unifiedresources/clone_test.go index 1bd583e334..078799b8eb 100644 --- a/internal/unifiedresources/clone_test.go +++ b/internal/unifiedresources/clone_test.go @@ -512,13 +512,19 @@ func TestCloneProxmoxData_NetworkInterfaceIsolation(t *testing.T) { } func TestCloneProxmoxDataMemoryIsolation(t *testing.T) { + observation := models.MemoryObservation{State: "last-known", Source: "guest-agent-meminfo", ObservedAt: time.Now().Add(-time.Minute)} original := &ProxmoxData{ - Memory: &models.Memory{Total: 8 << 30, Used: 4 << 30, Usage: 50}, + Memory: &models.Memory{Total: 8 << 30, Used: 4 << 30, Usage: 50, Observation: observation}, } cloned := cloneProxmoxData(original) + if cloned.Memory.Observation != observation { + t.Fatal("clone lost original memory observation") + } cloned.Memory.Used = 7 << 30 - if original.Memory.Used != 4<<30 { + cloned.Memory.Observation.State = "current" + cloned.Memory.Observation.ObservedAt = time.Now() + if original.Memory.Used != 4<<30 || original.Memory.Observation != observation { t.Error("mutating cloned memory should not affect original") } } diff --git a/internal/unifiedresources/metrics.go b/internal/unifiedresources/metrics.go index 039f0c4615..e691009e02 100644 --- a/internal/unifiedresources/metrics.go +++ b/internal/unifiedresources/metrics.go @@ -5,6 +5,7 @@ import ( "math" "math/rand" "strings" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/mockmode" "github.com/rcourtman/pulse-go-rewrite/internal/models" @@ -35,9 +36,19 @@ func hasObservedDiskUsage(disk models.Disk) bool { } func metricsFromHost(host models.Host) *ResourceMetrics { + // Author freshness from server-owned report receipt time, never from a + // guest row refreshed by another source or an agent-supplied annotation. + memory := host.Memory + memory.Observation = models.MemoryObservation{State: "current", Source: "agent", ObservedAt: host.LastSeen} + if host.Status != "online" || host.LastSeen.IsZero() || host.LastSeen.After(time.Now()) { + memory.Observation.State = "last-known" + if host.LastSeen.After(time.Now()) { + memory.Observation.ObservedAt = time.Time{} + } + } return buildHostMetricPayload( host.CPUUsage, - host.Memory, + memory, host.Disks, host.NetInRate, host.NetOutRate, @@ -99,7 +110,7 @@ func buildHostMetricPayload( metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source} if memory.Total > 0 && memory.HasKnownUsage() { percent := percentFromReportedPercent(memory.Usage) - metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source} + metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source, Observation: memory.Observation} } if len(disks) > 0 { disk := disks[0] @@ -199,7 +210,7 @@ func buildVMMetricPayload( metrics.CPU = &MetricValue{Value: cpuPercent, Percent: cpuPercent, Unit: "percent", Source: source} if memory.Total > 0 && memory.HasKnownUsage() { percent := percentFromUsage(memory.Usage) - metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source} + metrics.Memory = &MetricValue{Used: &memory.Used, Total: &memory.Total, Percent: percent, Unit: "bytes", Source: source, Observation: memory.Observation} } if hasObservedDiskUsage(disk) { percent := percentFromUsage(disk.Usage) diff --git a/internal/unifiedresources/metrics_test.go b/internal/unifiedresources/metrics_test.go index 99784a8e03..0e5616d115 100644 --- a/internal/unifiedresources/metrics_test.go +++ b/internal/unifiedresources/metrics_test.go @@ -544,3 +544,34 @@ func TestCanonicalDiskUsagePreservesUnavailableAndObservedZero(t *testing.T) { } } } + +func TestGuestMemoryMetricObservationFollowsSelectedSource(t *testing.T) { + at := time.Now().Add(-time.Minute) + retained := models.Memory{Total: 100, Used: 25, Free: 75, Usage: 25, + Observation: models.MemoryObservation{State: "last-known", Source: "guest-agent-meminfo", ObservedAt: at}} + vm, _ := resourceFromVM(models.VM{VMID: 105, Memory: retained}) + ct, _ := resourceFromContainer(models.Container{VMID: 106, Memory: retained}) + for _, resource := range []Resource{vm, ct} { + if resource.Metrics.Memory.Observation != retained.Observation || resource.Proxmox.Memory.Observation != retained.Observation { + t.Fatal("guest adapter lost selected/raw observation") + } + copy := cloneResource(&resource) + copy.Metrics.Memory.Observation.State = "current" + copy.Proxmox.Memory.Observation.Source = "changed" + if resource.Metrics.Memory.Observation != retained.Observation || resource.Proxmox.Memory.Observation != retained.Observation { + t.Fatal("clone aliases memory provenance") + } + } + // Missing platform usage must not mark a surviving live agent metric + // unavailable just because the retained platform facet says so. + unavailable := models.UnavailableMemory(100) + unavailable.Observation = models.MemoryObservation{State: "unavailable", Source: "unavailable"} + platform, _ := resourceFromVM(models.VM{VMID: 105, Memory: unavailable}) + agent := metricsFromHost(models.Host{Status: "online", LastSeen: at, + Memory: models.Memory{Total: 100, Used: 40, Free: 60, Usage: 40, + Observation: models.MemoryObservation{State: "current", Source: "untrusted", ObservedAt: time.Now()}}}) + merged := mergeMetrics(&platform, agent, platform.Metrics, SourceProxmox, time.Now(), nil, nil) + if merged.Memory == nil || merged.Memory.Observation.State != "current" || merged.Memory.Observation.Source != "agent" || !merged.Memory.Observation.ObservedAt.Equal(at) || *merged.Memory.Used != 40 { + t.Fatalf("cross-source selection lost agent receipt or trusts supplied provenance: %+v", merged.Memory) + } +} diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index ee5e4a1452..31b01a0e85 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -347,6 +347,9 @@ type MetricValue struct { Percent float64 `json:"percent,omitempty"` Unit string `json:"unit,omitempty"` Source DataSource `json:"-"` + // Memory provenance travels with the selected numeric metric, not a + // different platform facet retained on a cross-source merged guest. + Observation models.MemoryObservation `json:"observation,omitzero"` } // ProxmoxData contains Proxmox-specific data for a resource. diff --git a/internal/unifiedresources/views.go b/internal/unifiedresources/views.go index d6440bdff2..4219d0469e 100644 --- a/internal/unifiedresources/views.go +++ b/internal/unifiedresources/views.go @@ -88,7 +88,8 @@ func linkedAgentMemoryFromResource(r *Resource) (models.Memory, bool) { if r == nil || r.Agent == nil || r.Agent.Stale || r.Agent.Memory == nil { return models.Memory{}, false } - if status, ok := r.SourceStatus[SourceAgent]; !ok || status.Status != "online" { + status, ok := r.SourceStatus[SourceAgent] + if !ok || status.Status != "online" { return models.Memory{}, false } m := r.Agent.Memory @@ -97,6 +98,13 @@ func linkedAgentMemoryFromResource(r *Resource) (models.Memory, bool) { return models.Memory{}, false } memory.Usage = float64(m.Used) / float64(m.Total) * 100 + // Guest-row LastSeen can be renewed by PVE alone. The surviving agent + // sample keeps the agent source's timestamp after host/guest correlation. + memory.Observation = models.MemoryObservation{State: "current", Source: "agent", ObservedAt: status.LastSeen} + if status.LastSeen.IsZero() || status.LastSeen.After(time.Now()) { + memory.Observation.State = "last-known" + memory.Observation.ObservedAt = time.Time{} + } return memory, memory.HasKnownUsage() } diff --git a/internal/unifiedresources/views_test.go b/internal/unifiedresources/views_test.go index 2ec8712040..4840036535 100644 --- a/internal/unifiedresources/views_test.go +++ b/internal/unifiedresources/views_test.go @@ -12,6 +12,42 @@ import ( func ptrInt64(v int64) *int64 { return &v } func ptrInt(v int) *int { return &v } +func TestLinkedAgentMemoryObservationUsesItsOwnSourceTime(t *testing.T) { + observedAt := time.Now().Add(-time.Minute) + resource := &Resource{ + Type: ResourceTypeVM, LastSeen: time.Now(), + Agent: &AgentData{Memory: &AgentMemoryMeta{Total: 100, Used: 25, Free: 75}}, + SourceStatus: map[DataSource]SourceStatus{SourceAgent: {Status: "online", LastSeen: observedAt}}, + } + for _, guest := range []string{"qemu", "lxc"} { + t.Run(guest, func(t *testing.T) { + var memory models.Memory + var ok bool + if guest == "qemu" { + view := NewVMView(resource) + memory, ok = view.LinkedAgentMemory() + } else { + view := NewContainerView(resource) + memory, ok = view.LinkedAgentMemory() + } + if !ok || memory.Observation.State != "current" || memory.Observation.Source != "agent" || !memory.Observation.ObservedAt.Equal(observedAt) { + t.Fatalf("agent memory uses guest-row time or loses provenance: %+v", memory) + } + }) + } + for _, unknownTime := range []time.Time{{}, time.Now().Add(time.Hour)} { + resource.SourceStatus[SourceAgent] = SourceStatus{Status: "online", LastSeen: unknownTime} + memory, ok := linkedAgentMemoryFromResource(resource) + if !ok || memory.Observation.State != "last-known" || !memory.Observation.ObservedAt.IsZero() { + t.Fatalf("unknown/future receipt manufactured a current reading: %+v", memory) + } + } + resource.SourceStatus[SourceAgent] = SourceStatus{Status: "offline", LastSeen: observedAt} + if _, ok := linkedAgentMemoryFromResource(resource); ok { + t.Fatal("provenance made an offline agent sample selectable") + } +} + func assertStringSlice(t *testing.T, got, want []string) { t.Helper() if len(got) != len(want) {