From 1070667a6786275ca47a2396e6de0036e0a8481a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:33:19 +0100 Subject: [PATCH 1/2] Bound QEMU disk fallback to its original filesystem observation Retain recent filesystem-only evidence across safe deferrals without renewing its ten-minute age from VM polls or cached identity. Carry internal source provenance through canonical replacement, preserve independent linked-agent precedence and keep expired values unavailable until a new ordinary read succeeds. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 10 + .../v6/internal/subsystems/monitoring.md | 20 ++ .../internal/subsystems/storage-recovery.md | 9 + .../internal/subsystems/unified-resources.md | 17 ++ internal/models/metrics_types_test.go | 23 ++ internal/models/models.go | 9 + .../monitoring/canonical_guardrails_test.go | 7 + ...st_agent_optional_read_integration_test.go | 13 +- internal/monitoring/guest_disk_expiry_test.go | 283 ++++++++++++++++++ internal/monitoring/guest_disk_stability.go | 28 +- .../monitoring/guest_disk_stability_test.go | 1 + .../monitoring/monitor_host_agents_test.go | 1 + internal/monitoring/monitor_previous_state.go | 1 + .../monitoring/monitor_pve_guest_builders.go | 10 + internal/unifiedresources/adapters.go | 1 + .../canonical_identity_test.go | 18 ++ internal/unifiedresources/registry.go | 3 + .../registry_merge_policy_test.go | 27 ++ internal/unifiedresources/types.go | 7 +- internal/unifiedresources/views.go | 8 + internal/unifiedresources/views_test.go | 20 ++ 21 files changed, 505 insertions(+), 11 deletions(-) create mode 100644 internal/monitoring/guest_disk_expiry_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 724a5dfc7d..cf46293a2f 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -15,6 +15,16 @@ ## Purpose +### QGA disk provenance stays outside the agent report contract + +The shared VM model carries an internal filesystem source/time observation for +bounded QGA disk fallback. It is not agent input or a public JSON field; linked +Pulse-agent disk evidence keeps its own source and precedence. Agent report, +identity, enrollment and lifecycle contracts are unchanged. +`TestGuestDiskObservationIsInternalSnapshotEvidence` verifies internal copy +isolation and omission from both model and frontend JSON; the existing linked +agent exclusion control now supplies an explicit original QGA observation. + ### Automatic provider links retain full hostname distinctions An agent report may automatically associate a short hostname with a unique diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 525d568088..d3ef398cff 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,26 @@ ## Purpose +### QEMU disk fallbacks retain their original filesystem source time + +A successful QGA filesystem read owns the disk observation's source and original +receipt time. Internal model, canonical facet and previous-state copies preserve +that evidence. Repeated polling, cached OS/version identity and an optional +metadata failure cannot renew the existing ten-minute carry-forward window. +Recent disk-only readings remain eligible without optional metadata; expired, +future, missing or non-QGA origins do not. A legacy direct reading may use its +receipt once; an already unavailable/retained legacy value with unknown origin +cannot borrow the latest poll time. A new ordinary successful read restores the +observation, including measured zero. Linked Pulse-agent disks retain their own +source and precedence, not a fabricated QGA origin. Guest-command admission, +backup safety, History and alert missing-observation rules remain unchanged. + +`TestGuestDiskObservationContract` covers original/legacy age boundaries, +repeated canonical replacement, disk-only continuity, real-client deferral and +expiry in cluster/node collectors, cancelled detail, current zero recovery, +unchanged prior History and alert continuity. These are synthetic source checks, +not native thaw, QEMU-only Windows acceptance or completion of #2439/#2619. + ### Automatic Proxmox agent attribution preserves distinct FQDNs Hostname-only attribution uses the common full-hostname equivalence rule: diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index ee76d3b524..32fcabde93 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -21,6 +21,15 @@ ## Purpose +### Shared guest disk evidence is not recovery-point identity + +The canonical Proxmox facet carries internal QGA filesystem source/time evidence +for bounded monitoring fallback. It is omitted from JSON and does not change +guest identity, backup attribution, storage selection or recovery-point keys. +Complete guest observations replace a missing/expired origin rather than +inheriting it. Canonical disk observation wire/identity and merge controls cover +this shared type boundary; no native backup or recovery acceptance follows. + ### PMG collection scope remains separate from backup recovery PMG settings readback emits explicit false dataset options; absence must not diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 937effd895..8c0289a16f 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 +### Internal guest disk observation provenance + +A Proxmox VM facet preserves the successful filesystem read's original source +and time through adapter, clone, typed view and previous-guest projection. +Complete guest facets replace this evidence in both directions, including a +missing/expired origin; partial non-guest facets cannot renew or clear it. +Linked agent selection remains independent and cannot become QGA fallback +proof. `DiskObservation` is internal only (`json:"-"`): no JSON, identity, +History key, source-priority or agent report contract changes. + +`TestGuestDiskObservationDoesNotChangeWireOrIdentity`, +`TestGuestDiskObservationViewUsesOnlyItsOwnSource` and +`TestGuestDiskObservationMergeReplacesMissingAndDifferentSource` verify exact +public facet equality, internal cloning, missing-source access and authoritative +replacement. Monitoring's disk observation contract covers the ordinary poll +and served-reason boundary. These checks do not establish installed recovery. + ### Guest views retain exact governance identity for configuration reads VM and system-container views expose cloned canonical policy/AI-safe metadata diff --git a/internal/models/metrics_types_test.go b/internal/models/metrics_types_test.go index ee4c0a1554..9d7fc24629 100644 --- a/internal/models/metrics_types_test.go +++ b/internal/models/metrics_types_test.go @@ -7,6 +7,29 @@ import ( "time" ) +func TestGuestDiskObservationIsInternalSnapshotEvidence(t *testing.T) { + origin := GuestDiskObservation{Source: "guest-agent", ObservedAt: time.Now().Add(-time.Minute)} + vm := VM{ID: "pve:node:105", VMID: 105, Type: "qemu", DiskObservation: origin, + Disk: Disk{Total: 1000, Used: 400, Free: 600, Usage: 40}, Disks: []Disk{{Total: 1000, Used: 400, Usage: 40}}} + copy := cloneVMs([]VM{vm})[0] + if copy.DiskObservation != origin { + t.Fatal("snapshot cloning lost the filesystem's original source time") + } + copy.DiskObservation.ObservedAt = time.Now() + copy.Disks[0].Used = 999 + if vm.DiskObservation != origin || vm.Disks[0].Used != 400 { + t.Fatal("snapshot mutation changed source-owned evidence") + } + for name, value := range map[string]any{"model": vm, "frontend": vm.ToFrontend()} { + t.Run(name, func(t *testing.T) { + wire, err := json.Marshal(value) + if err != nil || strings.Contains(string(wire), "DiskObservation") || strings.Contains(string(wire), "diskObservation") || strings.Contains(string(wire), "ObservedAt") { + t.Fatalf("internal QGA evidence changed the public model/agent contract: %s / %v", wire, err) + } + }) + } +} + func TestGuestMemoryObservationWireContract(t *testing.T) { legacy := Memory{Total: 100, Used: 25, Free: 75, Usage: 25} payload, err := json.Marshal(legacy) diff --git a/internal/models/models.go b/internal/models/models.go index 941020da7d..604f44c486 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -261,6 +261,7 @@ type VM struct { Disk Disk `json:"disk"` Disks []Disk `json:"disks,omitempty"` DiskStatusReason string `json:"diskStatusReason,omitempty"` // Why disk stats are unavailable + DiskObservation GuestDiskObservation `json:"-"` // Source-owned carry-forward evidence, not the latest poll time. GuestAgentStatus string `json:"guestAgentStatus,omitempty"` // QEMU guest-agent runtime status GuestAgentExpected bool `json:"guestAgentExpected,omitempty"` IPAddresses []string `json:"ipAddresses,omitempty"` @@ -3228,6 +3229,14 @@ func (i GuestNetworkInterface) NormalizeCollections() GuestNetworkInterface { } // Disk represents disk usage +// GuestDiskObservation keeps a QEMU filesystem reading's original source time +// through internal snapshots. DiskStatusReason owns current/retained display +// state; a linked Pulse agent remains a separate source. This is not wire data. +type GuestDiskObservation struct { + Source string + ObservedAt time.Time +} + type Disk struct { Total int64 `json:"total"` Used int64 `json:"used"` diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 1b4d1cb454..a8c57fe652 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1124,6 +1124,13 @@ func TestProxmoxGuestDiskCarryForwardUsesCanonicalStabilityHelper(t *testing.T) } } +func TestGuestDiskObservationContract(t *testing.T) { + t.Run("legacy-origin-admission", testGuestDiskLegacyObservationAge) + t.Run("original-source-age", testGuestDiskOriginalObservationAge) + t.Run("repeated-canonical-polls", testGuestDiskRepeatedPollsAndCanonicalOrigin) + t.Run("ordinary-deferral-expiry-recovery", testGuestDiskOrdinaryDeferralExpiryAndRecovery) +} + func TestProxmoxGuestDiskInventoryPrefersCanonicalLinkedHostAgentSource(t *testing.T) { requiredSnippets := map[string][]string{ "guest_host_agent_fallback.go": { diff --git a/internal/monitoring/guest_agent_optional_read_integration_test.go b/internal/monitoring/guest_agent_optional_read_integration_test.go index 0aac05f645..6d8145a3f5 100644 --- a/internal/monitoring/guest_agent_optional_read_integration_test.go +++ b/internal/monitoring/guest_agent_optional_read_integration_test.go @@ -210,15 +210,10 @@ func testGuestAgentOptionalReadOrdering(t *testing.T, withoutStatus bool) { t.Fatal("completed unsupported reply prevented repeated fresh filesystem polls") } } else { - if withoutStatus || i == 2 { - if vm.Disk.Used != initial.Disk.Used || vm.DiskStatusReason != "prev-agent-cooldown" { - t.Fatalf("uncertainty lost eligible retained disk: %+v", vm.Disk) - } - } else if vm.Disk.Usage >= 0 || vm.DiskStatusReason != "agent-cooldown" { - // With no useful prior identity, a retained disk alone - // cannot repeatedly renew the evidence gate. Losing its - // display number is truthful unavailability, not zero. - t.Fatal("retained filesystem alone manufactured continuing guest evidence") + // Filesystem success owns its original age, independently + // of optional metadata support and the latest row refresh. + if vm.Disk.Used != initial.Disk.Used || vm.DiskStatusReason != "prev-agent-cooldown" || vm.DiskObservation != initial.DiskObservation { + t.Fatalf("uncertainty lost or renewed eligible retained disk: %+v", vm.Disk) } if vm.GuestAgentStatus != "deferred" { t.Fatal("unavailable/retained disk concealed active guest deferral") diff --git a/internal/monitoring/guest_disk_expiry_test.go b/internal/monitoring/guest_disk_expiry_test.go new file mode 100644 index 0000000000..8576f6987c --- /dev/null +++ b/internal/monitoring/guest_disk_expiry_test.go @@ -0,0 +1,283 @@ +package monitoring + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" +) + +// These assertions also compile over the parent: retained identity and the +// latest poll time cannot establish an unknown original filesystem read time. +func testGuestDiskLegacyObservationAge(t *testing.T) { + now := time.Date(2026, 10, 9, 14, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name, previousReason string + lastSeen time.Time + wantRetained bool + }{ + {"direct recent legacy observation", "", now.Add(-time.Minute), true}, + {"direct legacy at existing age boundary", "", now.Add(-recentGuestAgentEvidenceMaxAge), true}, + {"direct legacy expired", "", now.Add(-recentGuestAgentEvidenceMaxAge - time.Second), false}, + {"direct legacy missing time", "", time.Time{}, false}, + {"direct legacy future time", "", now.Add(time.Second), false}, + {"retained legacy missing original age", "prev-agent-error", now.Add(-time.Second), false}, + {"backup-retained legacy missing original age", "prev-vm-locked", now.Add(-time.Second), false}, + {"cooldown-retained legacy missing original age", "prev-agent-cooldown", now.Add(-time.Second), false}, + {"unavailable legacy with numeric carrier", "agent-error", now.Add(-time.Second), false}, + } { + t.Run(tc.name, func(t *testing.T) { + prev := &models.VM{Type: "qemu", Status: "running", LastSeen: tc.lastSeen, AgentVersion: "retained-version", DiskStatusReason: tc.previousReason, + Disk: models.Disk{Total: 1000, Used: 400, Free: 600, Usage: 40}, + Disks: []models.Disk{{Total: 1000, Used: 400, Free: 600, Usage: 40, Mountpoint: "/"}}} + total, used, free, usage, disks, reason := stabilizeGuestLowTrustDisk(prev, "running", 2000, 0, 2000, -1, nil, "agent-error", false, now) + if tc.wantRetained { + if total != 1000 || used != 400 || free != 600 || usage != 40 || len(disks) != 1 || reason != "prev-agent-error" { + t.Errorf("recent direct observation lost: %d/%d/%d/%v/%d/%s", total, used, free, usage, len(disks), reason) + } + } else if total != 2000 || used != 0 || free != 2000 || usage != -1 || len(disks) != 0 || reason != "agent-error" { + t.Errorf("unknown/expired original age became retained usage: %d/%d/%d/%v/%d/%s", total, used, free, usage, len(disks), reason) + } + }) + } +} + +func testGuestDiskOriginalObservationAge(t *testing.T) { + now := time.Date(2026, 10, 9, 14, 0, 0, 0, time.UTC) + for _, tc := range []struct { + name, source string + at time.Time + wantRetained bool + }{ + {"disk-only recent original", "guest-agent", now.Add(-time.Minute), true}, + {"disk-only inclusive original boundary", "guest-agent", now.Add(-recentGuestAgentEvidenceMaxAge), true}, + {"expired original despite fresh poll and identity", "guest-agent", now.Add(-recentGuestAgentEvidenceMaxAge - time.Nanosecond), false}, + {"future original", "guest-agent", now.Add(time.Second), false}, + {"missing original", "guest-agent", time.Time{}, false}, + {"unknown source", "unknown", now.Add(-time.Minute), false}, + {"missing source", "", now.Add(-time.Minute), false}, + {"linked agent is not QGA evidence", "agent", now.Add(-time.Minute), false}, + } { + t.Run(tc.name, func(t *testing.T) { + prev := &models.VM{Type: "qemu", Status: "running", LastSeen: now, DiskStatusReason: "prev-agent-error", + DiskObservation: models.GuestDiskObservation{Source: tc.source, ObservedAt: tc.at}, + Disk: models.Disk{Total: 1000, Used: 400, Free: 600, Usage: 40}, + Disks: []models.Disk{{Total: 1000, Used: 400, Free: 600, Usage: 40, Mountpoint: "/"}}} + if !tc.wantRetained { + prev.AgentVersion, prev.OSName = "retained-version", "retained-os" + } + _, used, _, usage, disks, reason := stabilizeGuestLowTrustDisk(prev, "running", 2000, 0, 2000, -1, nil, "vm-locked", false, now) + if tc.wantRetained { + if used != 400 || usage != 40 || len(disks) != 1 || reason != "prev-vm-locked" { + t.Errorf("original disk-only evidence lost: %d/%v/%d/%s", used, usage, len(disks), reason) + } + disks[0].Used = 999 + if prev.Disks[0].Used != 400 { + t.Fatal("retention aliased the previous disk inventory") + } + } else if used != 0 || usage != -1 || len(disks) != 0 || reason != "vm-locked" { + t.Errorf("expired/missing/wrong-source evidence became usage: %d/%v/%d/%s", used, usage, len(disks), reason) + } + if prev.DiskObservation.ObservedAt != tc.at { + t.Fatal("selector renewed the source time") + } + }) + } +} + +func testGuestDiskRepeatedPollsAndCanonicalOrigin(t *testing.T) { + original := time.Date(2026, 10, 9, 14, 0, 0, 0, time.UTC) + prev := models.VM{ID: "disk-expiry:node:105", Instance: "disk-expiry", Node: "node", VMID: 105, Type: "qemu", Status: "running", LastSeen: original, + DiskObservation: models.GuestDiskObservation{Source: "guest-agent", ObservedAt: original}, + Disk: models.Disk{Total: 1000, Used: 400, Free: 600, Usage: 40}, Disks: []models.Disk{{Total: 1000, Used: 400, Free: 600, Usage: 40, Mountpoint: "/"}}} + registry := unifiedresources.NewRegistry(nil) + for minute := 1; minute <= 12; minute++ { + t.Run(fmt.Sprintf("minute-%02d", minute), func(t *testing.T) { + now := original.Add(time.Duration(minute) * time.Minute) + total, used, free, usage, disks, reason := stabilizeGuestLowTrustDisk(&prev, "running", 2000, 0, 2000, -1, nil, "agent-error", false, now) + vm := prev + vm.LastSeen = now + vm.Disk = models.Disk{Total: int64(total), Used: int64(used), Free: int64(free), Usage: usage} + vm.Disks, vm.DiskStatusReason = disks, reason + vm.DiskObservation = models.GuestDiskObservation{Source: "guest-agent"} + if strings.HasPrefix(reason, "prev-") { + vm.DiskObservation.ObservedAt = guestDiskObservationTime(&prev) + } + wantTime := time.Time{} + if minute <= 10 { + wantTime = original + if used != 400 || usage != 40 || reason != "prev-agent-error" { + t.Errorf("disk-only evidence lost before expiry: %d/%v/%s", used, usage, reason) + } + } else if used != 0 || usage != -1 || reason != "agent-error" { + t.Errorf("minute %d renewed expired usage: %d/%v/%s", minute, used, usage, reason) + } + registry.IngestSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + view := registry.VMs()[0] + prev = previousVMFromView(view) + if prev.DiskObservation.Source != "guest-agent" || !prev.DiskObservation.ObservedAt.Equal(wantTime) || prev.DiskStatusReason != reason || !prev.LastSeen.Equal(now) { + t.Fatal("canonical replacement renewed, lost or resurrected the disk's original source time") + } + }) + } +} + +// Exercise ordinary HTTP reads and both guest collectors through the canonical +// previous-state boundary. Synthetic prior times avoid waiting ten minutes; +// no native backup/freeze, command failure injection or admission bypass occurs. +func testGuestDiskOrdinaryDeferralExpiryAndRecovery(t *testing.T) { + const mib = uint64(1024 * 1024) + for _, path := range []string{"cluster", "node"} { + for _, reason := range []string{"vm-locked", "lock-unverified"} { + t.Run(path+"/"+reason, func(t *testing.T) { + var phase, commands atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + if phase.Load() == 1 { + if reason == "vm-locked" { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } else { + fmt.Fprint(w, `{"data":null}`) + } + } else { + fmt.Fprint(w, `{"data":{}}`) + } + case strings.HasSuffix(r.URL.Path, "/status/current"): + fmt.Fprintf(w, `{"data":{"status":"running","agent":1,"maxmem":%d,"mem":%d,"meminfo":{"total":%d,"available":%d}}}`, 8*mib, 2*mib, 8*mib, 6*mib) + case strings.Contains(r.URL.Path, "/agent/"): + commands.Add(1) + if strings.HasSuffix(r.URL.Path, "/get-fsinfo") { + used := 980 * mib + if phase.Load() == 2 { + used = 0 // A measured zero is a successful recovery. + } + fmt.Fprintf(w, `{"data":{"result":[{"mountpoint":"C:\\","type":"ntfs","total-bytes":%d,"used-bytes":%d}]}}`, 1000*mib, used) + } else { + fmt.Fprint(w, `{"data":{"result":[]}}`) + } + default: + http.NotFound(w, r) + } + })) + defer server.Close() + client, err := proxmox.NewClient(proxmox.ClientConfig{Host: server.URL, TokenName: "fixture@pve!pulse", TokenValue: "fixture", Timeout: time.Second}) + if err != nil { + t.Fatal(err) + } + m := guestHistoryObservationMonitor(t) + m.alertManager = alerts.NewManagerWithDataDir(t.TempDir(), alerts.WithoutPersistedAlertRestore()) + t.Cleanup(m.alertManager.Stop) + alertConfig := m.alertManager.GetConfig() + alertConfig.Enabled = true + alertConfig.GuestDefaults.Disk = &alerts.HysteresisThreshold{Trigger: 90, Clear: 85} + alertConfig.MetricTimeThresholds = map[string]map[string]int{"guest": {"disk": 0}} + m.alertManager.UpdateConfig(alertConfig) + m.config, m.state, m.rateTracker = &config.Config{}, models.NewState(), NewRateTracker() + m.guestMetadataLimiter = make(map[string]time.Time) + registry := unifiedresources.NewRegistry(nil) + m.resourceStore = unifiedresources.NewMonitorAdapter(registry) + res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "windows", VMID: 105, Status: "running", MaxMem: 8 * mib, Mem: 2 * mib, MaxDisk: 1000 * mib, CPU: .2} + id := makeGuestID("disk-expiry", "node", 105) + publish := func(vm models.VM) { + registry.IngestSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + } + build := func(ctx context.Context) models.VM { + t.Helper() + previous := m.previousGuestContextForInstance("disk-expiry").vmsByID[id] + var vm models.VM + if path == "node" { + vms, _ := m.pollNodeVMsWithClusterResourceBuilder(ctx, "disk-expiry", "node", []proxmox.VM{{VMID: 105, Name: res.Name, Status: res.Status, MaxMem: res.MaxMem, Mem: res.Mem, MaxDisk: res.MaxDisk, CPU: res.CPU}}, client, map[string]models.VM{id: previous}, nil) + if len(vms) != 1 { + t.Fatal("node collector lost the guest") + } + vm = vms[0] + } else { + var ok bool + vm, _, _, _, _, ok = m.buildVMFromClusterResource(ctx, "disk-expiry", res, client, id, nil, &previous) + if !ok { + t.Fatal("cluster builder lost the guest") + } + m.checkGuestAlertsForVM("disk-expiry", vm) + } + publish(vm) + m.recordGuestMetrics([]models.VM{vm}, nil, time.Now().Add(-time.Second)) + view := registry.VMs()[0] + if view.DiskStatusReason() != vm.DiskStatusReason || view.DiskObservation() != vm.DiskObservation { + t.Fatal("source disk observation changed at canonical read boundary") + } + front := m.buildBroadcastFrontendStateFromSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}, m.mockModeFence.begin()) + wire, err := json.Marshal(front.Resources) + if err != nil || strings.Contains(string(wire), "DiskObservation") || strings.Contains(string(wire), "diskObservation") { + t.Fatal("internal carry-forward evidence leaked into public JSON") + } + var served []struct{ Proxmox unifiedresources.ProxmoxData } + if err := json.Unmarshal(wire, &served); err != nil || len(served) != 1 || served[0].Proxmox.DiskStatusReason != vm.DiskStatusReason { + t.Fatal("served source reason differs from the observed poll") + } + return vm + } + initial := build(context.Background()) + if initial.Disk.Used != int64(980*mib) || initial.DiskStatusReason != "" || initial.DiskObservation.Source != "guest-agent" || initial.DiskObservation.ObservedAt.IsZero() || initial.DiskObservation.ObservedAt.After(initial.LastSeen) { + t.Fatal("filesystem read did not establish its own original time") + } + originalHistory := guestHistoryStoredPoints(t, m, "vm", id, "disk") + if len(originalHistory) != 1 || originalHistory[0].Value != 98 || len(m.alertManager.GetActiveAlerts()) != 1 { + t.Fatal("current filesystem evidence failed to establish History and an alert") + } + before := commands.Load() + phase.Store(1) + for poll := 0; poll < 3; poll++ { + retained := build(context.Background()) + if retained.Disk.Used != initial.Disk.Used || retained.DiskStatusReason != "prev-"+reason || retained.DiskObservation != initial.DiskObservation || len(retained.Disks) != 1 { + t.Fatal("disk-only last-known value or original source time lost during deferral") + } + } + // Shift only the accepted source's observation time, not LastSeen. + // This models an extended pause without an actual ten-minute delay. + aged := m.previousGuestContextForInstance("disk-expiry").vmsByID[id] + aged.DiskObservation.ObservedAt = time.Now().Add(-recentGuestAgentEvidenceMaxAge - time.Second) + publish(aged) + expired := build(context.Background()) + if expired.Disk.Usage != -1 || expired.DiskStatusReason != reason || !expired.DiskObservation.ObservedAt.IsZero() || len(expired.Disks) != 0 || expired.GuestAgentStatus != "deferred" { + t.Fatal("extended pause manufactured retained usage or hid the admission deferral") + } + // Exhausted detail budget may retain text inventory, but must not + // restore the expired numeric summary or a successful read time. + canceled, cancel := context.WithCancel(context.Background()) + cancel() + budgeted := build(canceled) + if strings.HasPrefix(budgeted.DiskStatusReason, "prev-") || !budgeted.DiskObservation.ObservedAt.IsZero() { + t.Fatal("canceled enrichment revived expired disk evidence") + } + if commands.Load() != before || !reflect.DeepEqual(guestHistoryStoredPoints(t, m, "vm", id, "disk"), originalHistory) || len(m.metricsHistory.GetGuestMetrics(id, "disk", time.Hour)) != 1 || len(m.metricsHistory.GetGuestMetrics(id, "cpu", time.Hour)) != 6 || len(m.alertManager.GetActiveAlerts()) != 1 { + t.Fatal("deferral dispatched a guest command, renewed disk History or suppressed independent CPU History") + } + phase.Store(2) + // Persisted History keys have second precision; make recovery + // distinct so zero cannot overwrite the original trusted point. + time.Sleep(1100 * time.Millisecond) + resumed := build(context.Background()) + if resumed.Disk.Usage != 0 || resumed.Disk.Used != 0 || resumed.DiskStatusReason != "" || resumed.DiskObservation.ObservedAt.IsZero() || !resumed.DiskObservation.ObservedAt.After(initial.DiskObservation.ObservedAt) || commands.Load() <= before || len(m.alertManager.GetActiveAlerts()) != 0 { + t.Fatal("new ordinary filesystem read failed to recover with measured zero") + } + points := guestHistoryStoredPoints(t, m, "vm", id, "disk") + if len(points) != 2 || points[0] != originalHistory[0] || points[1].Value != 0 { + t.Fatal("ordinary recovery lost original History or failed to append measured zero") + } + }) + } + } +} diff --git a/internal/monitoring/guest_disk_stability.go b/internal/monitoring/guest_disk_stability.go index 988020d128..9a8451f5aa 100644 --- a/internal/monitoring/guest_disk_stability.go +++ b/internal/monitoring/guest_disk_stability.go @@ -63,7 +63,7 @@ func stabilizeGuestLowTrustDisk( if status != "running" || diskFromAgent || !shouldCarryForwardQEMUDisk(diskStatusReason) { return diskTotal, diskUsed, diskFree, diskUsage, individualDisks, diskStatusReason } - if prev == nil || prev.Type != "qemu" || !hasRecentGuestAgentEvidence(prev, now) { + if prev == nil || prev.Type != "qemu" || !guestDiskSnapshotWithinAge(prev, now) { return diskTotal, diskUsed, diskFree, diskUsage, individualDisks, diskStatusReason } if prev.Disk.Total <= 0 || prev.Disk.Used < 0 || prev.Disk.Used > prev.Disk.Total || prev.Disk.Usage < 0 { @@ -79,3 +79,29 @@ func stabilizeGuestLowTrustDisk( return total, used, free, prev.Disk.Usage, cloneGuestDisks(prev.Disks), "prev-" + diskStatusReason } + +// Disk evidence belongs to the successful filesystem read, independently of +// cached identity or optional OS/version support. A legacy direct observation +// may use its receipt time once, but an already unavailable/retained value with +// no original time cannot borrow the latest VM poll's LastSeen. +func guestDiskObservationTime(prev *models.VM) time.Time { + if prev == nil { + return time.Time{} + } + observation := prev.DiskObservation + if observation.Source != "" || !observation.ObservedAt.IsZero() { + if observation.Source != "guest-agent" { + return time.Time{} + } + return observation.ObservedAt + } + if prev.DiskStatusReason != "" { + return time.Time{} + } + return prev.LastSeen +} + +func guestDiskSnapshotWithinAge(prev *models.VM, now time.Time) bool { + observedAt := guestDiskObservationTime(prev) + return !observedAt.IsZero() && !observedAt.After(now) && now.Sub(observedAt) <= recentGuestAgentEvidenceMaxAge +} diff --git a/internal/monitoring/guest_disk_stability_test.go b/internal/monitoring/guest_disk_stability_test.go index f449e9f538..c51124530a 100644 --- a/internal/monitoring/guest_disk_stability_test.go +++ b/internal/monitoring/guest_disk_stability_test.go @@ -102,6 +102,7 @@ func TestStabilizeGuestLowTrustDiskCarriesPreviouslyForwardedSnapshotWithAgentEv LastSeen: now.Add(-time.Minute), AgentVersion: "8.2.0", DiskStatusReason: "prev-no-filesystems", + DiskObservation: models.GuestDiskObservation{Source: "guest-agent", ObservedAt: now.Add(-time.Minute)}, Disk: models.Disk{ Total: 1000, Used: 400, diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index e2919b8a8f..b4127d27bf 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -7706,6 +7706,7 @@ func TestPreviousVMFromViewKeepsLinkedAgentDiskOutOfProxmoxCarry(t *testing.T) { VMID: 101, RuntimeStatus: "running", DiskStatusReason: "agent-not-running", + DiskObservation: models.GuestDiskObservation{Source: "guest-agent", ObservedAt: now.Add(-time.Minute)}, }, // The linked agent's addresses count as recent guest evidence. Identity: unifiedresources.ResourceIdentity{IPAddresses: []string{"10.0.0.5"}}, diff --git a/internal/monitoring/monitor_previous_state.go b/internal/monitoring/monitor_previous_state.go index 2ddb1449a9..c286b4a92b 100644 --- a/internal/monitoring/monitor_previous_state.go +++ b/internal/monitoring/monitor_previous_state.go @@ -163,6 +163,7 @@ func previousVMFromView(vm *unifiedresources.VMView) models.VM { NetworkInterfaces: guestNetworkInterfacesFromReadStateView(vm.NetworkInterfaces()), Disks: guestDisksFromReadStateView(vm.Disks()), DiskStatusReason: vm.DiskStatusReason(), + DiskObservation: vm.DiskObservation(), LastSeen: vm.LastSeen(), } } diff --git a/internal/monitoring/monitor_pve_guest_builders.go b/internal/monitoring/monitor_pve_guest_builders.go index e3ed5e553d..70e5c03581 100644 --- a/internal/monitoring/monitor_pve_guest_builders.go +++ b/internal/monitoring/monitor_pve_guest_builders.go @@ -32,6 +32,7 @@ type vmBuildState struct { diskUsage float64 diskFromAgent bool diskStatusReason string + diskObservation models.GuestDiskObservation guestAgentStatus string guestAgentExpected bool individualDisks []models.Disk @@ -143,6 +144,9 @@ func (m *Monitor) applyVMGuestAgentFSInfo(ctx context.Context, instanceName stri if len(fsDisks) > 0 { state.individualDisks = fsDisks } + if state.diskFromAgent { + state.diskObservation.ObservedAt = time.Now() + } if guestAgentDiskDeferred(state.diskStatusReason) { state.guestAgentStatus = "deferred" } @@ -226,6 +230,7 @@ func (m *Monitor) buildVMFromClusterResource( m.hasRecentGuestMetadataEvidence(instanceName, res.Node, res.VMID, prePollTime) state := vmBuildState{ + diskObservation: models.GuestDiskObservation{Source: "guest-agent"}, memTotal: res.MaxMem, memUsed: res.Mem, memorySource: "cluster-resources", @@ -355,6 +360,7 @@ func (m *Monitor) buildVMFromClusterResource( state.diskStatusReason, ) if preferred { + state.diskObservation = models.GuestDiskObservation{Source: "agent", ObservedAt: vmIDToHostAgent[guestID].LastSeen} log.Debug(). Str("instance", instanceName). Str("vm", res.Name). @@ -378,6 +384,9 @@ func (m *Monitor) buildVMFromClusterResource( state.diskFromAgent, sampleTime, ) + if strings.HasPrefix(state.diskStatusReason, "prev-") { + state.diskObservation.ObservedAt = guestDiskObservationTime(prevVM) + } if res.Status != "running" { state.memorySource = "powered-off" @@ -493,6 +502,7 @@ func (m *Monitor) buildVMFromClusterResource( }, Disks: state.individualDisks, DiskStatusReason: state.diskStatusReason, + DiskObservation: state.diskObservation, GuestAgentStatus: state.guestAgentStatus, Lock: res.Lock, GuestAgentExpected: state.guestAgentExpected, diff --git a/internal/unifiedresources/adapters.go b/internal/unifiedresources/adapters.go index 8c8cfd3f70..03e7328696 100644 --- a/internal/unifiedresources/adapters.go +++ b/internal/unifiedresources/adapters.go @@ -1843,6 +1843,7 @@ func resourceFromVM(vm models.VM) (Resource, ResourceIdentity) { LastBackup: vm.LastBackup, BackupInProgress: vm.BackupInProgress, DiskStatusReason: vm.DiskStatusReason, + DiskObservation: vm.DiskObservation, GuestAgentStatus: vm.GuestAgentStatus, GuestAgentExpected: vm.GuestAgentExpected, OSName: vm.OSName, diff --git a/internal/unifiedresources/canonical_identity_test.go b/internal/unifiedresources/canonical_identity_test.go index ad7f95816d..37e1b82581 100644 --- a/internal/unifiedresources/canonical_identity_test.go +++ b/internal/unifiedresources/canonical_identity_test.go @@ -5,10 +5,28 @@ import ( "reflect" "strings" "testing" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/models" ) +func TestGuestDiskObservationDoesNotChangeWireOrIdentity(t *testing.T) { + vm := models.VM{ID: "pve:node:105", VMID: 105, Name: "guest", Instance: "pve", Node: "node", Type: "qemu", Status: "running", LastSeen: time.Now(), + Disk: models.Disk{Total: 1000, Used: 400, Usage: 40}, + DiskObservation: models.GuestDiskObservation{Source: "guest-agent", ObservedAt: time.Now().Add(-time.Minute)}} + resource, identity := resourceFromVM(vm) + wire, err := json.Marshal(resource.Proxmox) + if err != nil || strings.Contains(string(wire), "Observation") || strings.Contains(string(wire), "observedAt") { + t.Fatalf("internal disk evidence changed the served facet: %s / %v", wire, err) + } + vm.DiskObservation = models.GuestDiskObservation{} + legacyResource, legacyIdentity := resourceFromVM(vm) + legacyWire, err := json.Marshal(legacyResource.Proxmox) + if err != nil || string(wire) != string(legacyWire) || !reflect.DeepEqual(identity, legacyIdentity) { + t.Fatal("internal disk evidence changed guest identity or any public facet byte") + } +} + func TestRefreshCanonicalIdentityPrefersTargetsAndCanonicalHostData(t *testing.T) { resource := Resource{ ID: "agent-1", diff --git a/internal/unifiedresources/registry.go b/internal/unifiedresources/registry.go index 1b6b6837ae..46f6a237e1 100644 --- a/internal/unifiedresources/registry.go +++ b/internal/unifiedresources/registry.go @@ -4242,6 +4242,9 @@ func mergeProxmoxData(existing *ProxmoxData, incoming *ProxmoxData) *ProxmoxData // leave a recovered guest paused forever. Non-guest partial facets do // not own these fields. merged.DiskStatusReason = incoming.DiskStatusReason + // A completed guest observation also owns loss/recovery of its original + // filesystem evidence. Missing time must not inherit an older origin. + merged.DiskObservation = incoming.DiskObservation merged.GuestAgentStatus = incoming.GuestAgentStatus merged.GuestAgentExpected = incoming.GuestAgentExpected } diff --git a/internal/unifiedresources/registry_merge_policy_test.go b/internal/unifiedresources/registry_merge_policy_test.go index 40dc5032e0..80dbe8cf3b 100644 --- a/internal/unifiedresources/registry_merge_policy_test.go +++ b/internal/unifiedresources/registry_merge_policy_test.go @@ -14,6 +14,33 @@ import ( "github.com/rcourtman/pulse-go-rewrite/pkg/diskinventory" ) +func TestGuestDiskObservationMergeReplacesMissingAndDifferentSource(t *testing.T) { + original := models.GuestDiskObservation{Source: "guest-agent", ObservedAt: time.Now().Add(-time.Minute)} + existing := &ProxmoxData{VMID: 105, DiskObservation: original, DiskStatusReason: "prev-agent-error"} + for _, tc := range []struct { + name string + incoming ProxmoxData + want models.GuestDiskObservation + }{ + {"expired original clears", ProxmoxData{VMID: 105, DiskObservation: models.GuestDiskObservation{Source: "guest-agent"}}, models.GuestDiskObservation{Source: "guest-agent"}}, + {"missing full guest origin clears", ProxmoxData{VMID: 105}, models.GuestDiskObservation{}}, + {"current linked source does not inherit QGA", ProxmoxData{VMID: 105, DiskObservation: models.GuestDiskObservation{Source: "agent", ObservedAt: time.Now()}}, models.GuestDiskObservation{Source: "agent"}}, + {"new successful filesystem read replaces", ProxmoxData{VMID: 105, DiskObservation: models.GuestDiskObservation{Source: "guest-agent", ObservedAt: time.Now()}}, models.GuestDiskObservation{Source: "guest-agent"}}, + {"non-guest partial facet does not own origin", ProxmoxData{NodeDisplayName: "node"}, original}, + } { + t.Run(tc.name, func(t *testing.T) { + want := tc.want + if !tc.incoming.DiskObservation.ObservedAt.IsZero() { + want.ObservedAt = tc.incoming.DiskObservation.ObservedAt + } + merged := mergeProxmoxData(existing, &tc.incoming) + if merged.DiskObservation != want || existing.DiskObservation != original { + t.Fatal("merge renewed/resurrected old evidence or mutated its source") + } + }) + } +} + func TestLinkedMergeAllowsOneSidedNodeHostLinkWhenHostnameCorroborates(t *testing.T) { registry := NewRegistry(NewMemoryStore()) diff --git a/internal/unifiedresources/types.go b/internal/unifiedresources/types.go index 6a049d8374..31e05682d0 100644 --- a/internal/unifiedresources/types.go +++ b/internal/unifiedresources/types.go @@ -423,7 +423,12 @@ type ProxmoxData struct { // Reclaimable buff/cache split out of the memory metric's free bytes. MemoryCache int64 `json:"memoryCache,omitempty"` Memory *models.Memory `json:"memory,omitempty"` - Lock string `json:"lock,omitempty"` // Proxmox lock state (e.g. "backup", "migrate", "snapshot") + + // Original QGA filesystem evidence survives internal clone/merge/previous + // state, never JSON. Source identity or UpdatedAt cannot renew it. + DiskObservation models.GuestDiskObservation `json:"-"` + + Lock string `json:"lock,omitempty"` // Proxmox lock state (e.g. "backup", "migrate", "snapshot") // SensorSetupOutdated is derived by the registry, never ingested: the // node's SSH temperature monitoring still runs the pre-rc.6 setup and a // disk under it waits on a SMART temperature that setup cannot deliver diff --git a/internal/unifiedresources/views.go b/internal/unifiedresources/views.go index e452c79483..2a8196faaf 100644 --- a/internal/unifiedresources/views.go +++ b/internal/unifiedresources/views.go @@ -259,6 +259,14 @@ func (v VMView) DiskStatusReason() string { return v.r.Proxmox.DiskStatusReason } +// DiskObservation is internal source evidence, not resource freshness. +func (v VMView) DiskObservation() models.GuestDiskObservation { + if v.r == nil || v.r.Proxmox == nil { + return models.GuestDiskObservation{} + } + return v.r.Proxmox.DiskObservation +} + func (v VMView) OSName() string { if v.r == nil || v.r.Proxmox == nil { return "" diff --git a/internal/unifiedresources/views_test.go b/internal/unifiedresources/views_test.go index e5e330402e..f2b464fc0f 100644 --- a/internal/unifiedresources/views_test.go +++ b/internal/unifiedresources/views_test.go @@ -9,6 +9,26 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/storagehealth" ) +func TestGuestDiskObservationViewUsesOnlyItsOwnSource(t *testing.T) { + origin := models.GuestDiskObservation{Source: "guest-agent", ObservedAt: time.Now().Add(-time.Minute)} + vm := models.VM{VMID: 105, Type: "qemu", Status: "running", LastSeen: time.Now(), DiskObservation: origin} + resource, _ := resourceFromVM(vm) + view := VMView{r: &resource} + if view.DiskObservation() != origin { + t.Fatal("adapter/view replaced filesystem source time with the row's receipt time") + } + copy := cloneResource(&resource) + copy.Proxmox.DiskObservation.ObservedAt = time.Now() + if view.DiskObservation() != origin { + t.Fatal("cloned resource observation mutated the original view") + } + for _, empty := range []VMView{{}, {r: &Resource{}}} { + if empty.DiskObservation() != (models.GuestDiskObservation{}) { + t.Fatal("missing source manufactured filesystem evidence") + } + } +} + func ptrInt64(v int64) *int64 { return &v } func ptrInt(v int) *int { return &v } From 32342c32c9d68472eb531d10305a978b4dcb8ae3 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Fri, 9 Oct 2026 15:41:11 +0100 Subject: [PATCH 2/2] Exercise accepted poll publication in disk fallback lifecycle proof Publish the synthetic guest generation through the ordinary state/store ingest boundary before checking served JSON. Read hydration may reuse a recent generation and is not poll acceptance. Preserve the initial four wire assertion failures and the unchanged regression assertions; fix the adjacent shared-model GoDoc attachment. Contract-Neutral: Fixture publication and comment attachment only; runtime behaviour, source provenance and public contracts are unchanged. Change-source: pulse-maintainer --- internal/models/models.go | 2 +- internal/monitoring/guest_disk_expiry_test.go | 9 ++++++--- 2 files changed, 7 insertions(+), 4 deletions(-) diff --git a/internal/models/models.go b/internal/models/models.go index 604f44c486..2cac70203e 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -3228,7 +3228,6 @@ func (i GuestNetworkInterface) NormalizeCollections() GuestNetworkInterface { return i } -// Disk represents disk usage // GuestDiskObservation keeps a QEMU filesystem reading's original source time // through internal snapshots. DiskStatusReason owns current/retained display // state; a linked Pulse agent remains a separate source. This is not wire data. @@ -3237,6 +3236,7 @@ type GuestDiskObservation struct { ObservedAt time.Time } +// Disk represents disk usage type Disk struct { Total int64 `json:"total"` Used int64 `json:"used"` diff --git a/internal/monitoring/guest_disk_expiry_test.go b/internal/monitoring/guest_disk_expiry_test.go index 8576f6987c..53c860aa51 100644 --- a/internal/monitoring/guest_disk_expiry_test.go +++ b/internal/monitoring/guest_disk_expiry_test.go @@ -192,7 +192,10 @@ func testGuestDiskOrdinaryDeferralExpiryAndRecovery(t *testing.T) { res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "windows", VMID: 105, Status: "running", MaxMem: 8 * mib, Mem: 2 * mib, MaxDisk: 1000 * mib, CPU: .2} id := makeGuestID("disk-expiry", "node", 105) publish := func(vm models.VM) { - registry.IngestSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + // An accepted poll owns publication. Read hydration may reuse a + // recent generation and is not an ingestion boundary. + m.state.UpdateVMs([]models.VM{vm}) + m.updateResourceStore(m.currentStateWithScope()) } build := func(ctx context.Context) models.VM { t.Helper() @@ -214,7 +217,7 @@ func testGuestDiskOrdinaryDeferralExpiryAndRecovery(t *testing.T) { } publish(vm) m.recordGuestMetrics([]models.VM{vm}, nil, time.Now().Add(-time.Second)) - view := registry.VMs()[0] + view := m.currentModeReadState().VMs()[0] if view.DiskStatusReason() != vm.DiskStatusReason || view.DiskObservation() != vm.DiskObservation { t.Fatal("source disk observation changed at canonical read boundary") } @@ -225,7 +228,7 @@ func testGuestDiskOrdinaryDeferralExpiryAndRecovery(t *testing.T) { } var served []struct{ Proxmox unifiedresources.ProxmoxData } if err := json.Unmarshal(wire, &served); err != nil || len(served) != 1 || served[0].Proxmox.DiskStatusReason != vm.DiskStatusReason { - t.Fatal("served source reason differs from the observed poll") + t.Fatalf("served source reason differs from poll %q: %s / %v", vm.DiskStatusReason, wire, err) } return vm }