diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 3c89b76df0..7d9fc4f39e 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -15,6 +15,35 @@ ## Purpose +### Retained guest observations are not alert evidence + +Guest memory marked `last-known`, `unavailable` or an unknown observation state +cannot start, advance, refresh or resolve a memory incident. A VM filesystem +read with a nonempty `diskStatusReason`, including retained `prev-*` values, +or unavailable aggregate usage is likewise not current disk evidence. It cannot +evaluate either aggregate or per-filesystem thresholds, replace an aggregate +incident with filesystem incidents, or clear filesystem alerts as if a failed +or expired inventory were authoritative removal. Metric filters do not match +these unavailable memory/disk values, including a fabricated healthy zero. + +Existing incidents retain their last trusted value, occurrence, acknowledgement, +observation time and lifecycle/delivery evidence until a current observation +resumes evaluation. Successful current inventory still resolves removed +filesystems. Explicit disablement, suppression and stopped-guest behavior remain +effective independently of telemetry, including per-filesystem disable overrides +on retained identities. Legacy unannotated memory and independent +current PVE/Pulse-agent memory/disks keep their existing policy, even while QGA +is deferred; current CPU and I/O continue to evaluate. Producers own freshness; +no new age threshold, QGA request, payload, alert family or UI surface is added. + +`guest_observation_test.go` pins typed/pointer VM/LXC observation admission, +delayed breaches, canonical incident/acknowledgement/ledger continuity, filters, +fresh recovery and removal, independent metrics and deliberate disablement. +`TestGuestBackupDeferralAlertObservationLifecycle` connects actual HTTP client, +poll builder, retained snapshot, unified view, History and the production alert +call. These are synthetic source proofs, not native backup/thaw, containing +release, installed delivery or external recipient acceptance. + ### Pulse system alerts retain their service scope in the existing UI The `pulse-system-` identity prefix or explicit `systemAlert: true` metadata diff --git a/internal/alerts/filter_evaluation.go b/internal/alerts/filter_evaluation.go index 5741e176a1..d8c0954130 100644 --- a/internal/alerts/filter_evaluation.go +++ b/internal/alerts/filter_evaluation.go @@ -19,18 +19,20 @@ func (m *Manager) evaluateFilterCondition(guest any, condition FilterCondition) // guestMetrics holds common metrics for filter evaluation type guestMetrics struct { - CPU float64 // CPU usage as percentage (0-100) - MemUsage float64 // Memory usage percentage - DiskUsage float64 // Disk usage percentage - DiskRead int64 // Bytes/s - DiskWrite int64 // Bytes/s - NetworkIn int64 // Bytes/s - NetworkOut int64 // Bytes/s - IORateValidity models.IORateValidity - Name string - Node string - ID string - Status string + CPU float64 // CPU usage as percentage (0-100) + MemUsage float64 // Memory usage percentage + DiskUsage float64 // Disk usage percentage + MemoryUnavailable bool + DiskUnavailable bool + DiskRead int64 // Bytes/s + DiskWrite int64 // Bytes/s + NetworkIn int64 // Bytes/s + NetworkOut int64 // Bytes/s + IORateValidity models.IORateValidity + Name string + Node string + ID string + Status string } // extractGuestMetrics extracts common metrics from a VM or Container @@ -57,8 +59,14 @@ func evaluateGuestCondition(metrics guestMetrics, condition FilterCondition) boo case "cpu": value = metrics.CPU case "memory": + if metrics.MemoryUnavailable { + return false + } value = metrics.MemUsage case "disk": + if metrics.DiskUnavailable { + return false + } value = metrics.DiskUsage case "diskread": if !validity.DiskRead { diff --git a/internal/alerts/guest.go b/internal/alerts/guest.go index b185ea1d30..25963829ef 100644 --- a/internal/alerts/guest.go +++ b/internal/alerts/guest.go @@ -259,8 +259,11 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { memoryMetric = &UnifiedResourceMetric{Percent: memUsage} } diskReadMetric, diskWriteMetric, networkInMetric, networkOutMetric := guestIORateMetrics(snapshot) - diskMetric := &UnifiedResourceMetric{Percent: diskUsage} - if len(disks) > 0 { + var diskMetric *UnifiedResourceMetric + if !snapshot.DiskUnavailable { + diskMetric = &UnifiedResourceMetric{Percent: diskUsage} + } + if !snapshot.DiskUnavailable && len(disks) > 0 { seenAggregateDiskKeys := make(map[string]struct{}) for idx, disk := range disks { if disk.Total <= 0 || disk.Usage < 0 { @@ -295,6 +298,12 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { NetworkOut: networkOutMetric, }, thresholds, evalOpts) + // Retained/failed filesystem evidence is useful to display, but neither + // its values nor an empty result can establish a breach, recovery or disk + // removal. CPU/I/O and independent current memory were evaluated above. + // Still apply explicit per-filesystem disablement to retained identities; + // this is operator policy, not recovery inferred from an old value. + if len(disks) > 0 { seenDiskKeys := make(map[string]struct{}) seenDiskResources := make(map[string]struct{}) @@ -331,6 +340,12 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { effectiveDiskThreshold = ensureHysteresisThreshold(diskOverride.Disk) } } + if snapshot.DiskUnavailable { + if effectiveDiskThreshold == nil || effectiveDiskThreshold.Trigger <= 0 { + m.clearAlert(canonicalMetricStateID(perDiskResourceID, "disk")) + } + continue + } log.Debug(). Str("guest", name). @@ -376,11 +391,15 @@ func (m *Manager) CheckGuest(guest any, instanceName string) { MonitorOnly: monitorOnly, }) } - if cleared := m.cleanupGuestDiskAlerts(guestID, seenDiskResources); cleared > 0 { - m.saveActiveAlertsAsync("guest-disk-set-changed") + if !snapshot.DiskUnavailable { + if cleared := m.cleanupGuestDiskAlerts(guestID, seenDiskResources); cleared > 0 { + m.saveActiveAlertsAsync("guest-disk-set-changed") + } + } + } else if !snapshot.DiskUnavailable { + if cleared := m.cleanupGuestDiskAlerts(guestID, nil); cleared > 0 { + m.saveActiveAlertsAsync("guest-disk-alerts-cleared") } - } else if cleared := m.cleanupGuestDiskAlerts(guestID, nil); cleared > 0 { - m.saveActiveAlertsAsync("guest-disk-alerts-cleared") } } diff --git a/internal/alerts/guest_observation_test.go b/internal/alerts/guest_observation_test.go new file mode 100644 index 0000000000..d231e727a3 --- /dev/null +++ b/internal/alerts/guest_observation_test.go @@ -0,0 +1,220 @@ +package alerts + +import ( + "bytes" + "encoding/json" + "sort" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog" + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +func guestObservationManager(t *testing.T) *Manager { + t.Helper() + m := newEventLogManager(t) + cfg := characterizationBaseConfig() + cfg.GuestDefaults.DiskWrite = &HysteresisThreshold{Trigger: 5, Clear: 4} + m.UpdateConfig(cfg) + disableTestTimeThresholds(m) + return m +} + +func guestObservationVM() models.VM { + return models.VM{ + ID: "site:node:105", VMID: 105, Name: "database", Node: "node", Instance: "site", Type: "qemu", Status: "running", + Memory: models.Memory{Total: 1000, Used: 960, Free: 40, Usage: 96, + Observation: models.MemoryObservation{State: "current", Source: "guest-agent-meminfo", ObservedAt: time.Now().Add(-time.Second)}}, + Disk: models.Disk{Total: 1000, Used: 980, Free: 20, Usage: 98}, + Disks: []models.Disk{{Total: 1000, Used: 980, Free: 20, Usage: 98, Mountpoint: "/", Device: "/dev/vda"}, + {Total: 1000, Used: 980, Free: 20, Usage: 98, Mountpoint: "/data", Device: "/dev/vdb"}}, + IORateValidity: models.IORateValidity{Explicit: true}, + } +} + +func guestObservationBytes(t *testing.T, m *Manager) []byte { + t.Helper() + active := m.GetActiveAlerts() + sort.Slice(active, func(i, j int) bool { return active[i].ID < active[j].ID }) + data, err := json.Marshal(active) + if err != nil { + t.Fatal(err) + } + return data +} + +func TestGuestAlertRetainedObservationsCannotFire(t *testing.T) { + for _, state := range []string{"last-known", "unavailable", "unknown-version"} { + for _, kind := range []string{"vm", "vm-pointer", "lxc", "lxc-pointer"} { + t.Run(kind+"/"+state, func(t *testing.T) { + m := guestObservationManager(t) + vm := guestObservationVM() + vm.Memory.Observation.State = state + vm.DiskStatusReason, vm.GuestAgentStatus = "prev-vm-locked", "deferred" + var guest any = vm + switch kind { + case "vm-pointer": + guest = &vm + case "lxc", "lxc-pointer": + ct := models.Container{ID: vm.ID, VMID: vm.VMID, Name: vm.Name, Node: vm.Node, Instance: vm.Instance, Status: vm.Status, Memory: vm.Memory, + Disk: models.Disk{Usage: 10}, IORateValidity: vm.IORateValidity} + guest = ct + if kind == "lxc-pointer" { + guest = &ct + } + } + for poll := 0; poll < 4; poll++ { + m.CheckGuest(guest, "site") + } + if got := m.GetActiveAlerts(); len(got) != 0 { + t.Fatalf("retained observations opened alerts: %+v", got) + } + if len(queryAlertEvents(t, m, eventlog.Filter{})) != 0 { + t.Fatal("retained observations emitted lifecycle/delivery evidence") + } + }) + } + } +} + +func TestGuestAlertUnavailableDiskInventoryIsNotRecovery(t *testing.T) { + for _, reason := range []string{"prev-vm-locked", "prev-lock-unverified", "agent-timeout", "agent-not-running", "permission-denied", "agent-error", "no-filesystems"} { + t.Run(reason, func(t *testing.T) { + m := guestObservationManager(t) + vm := guestObservationVM() + m.CheckGuest(vm, "site") + active := m.GetActiveAlerts() + if len(active) != 3 { + t.Fatalf("current memory and two filesystems must fire: %+v", active) + } + for _, a := range active { + if err := m.AcknowledgeAlert(a.ID, "operator"); err != nil { + t.Fatal(err) + } + } + before := guestObservationBytes(t, m) + events := len(queryAlertEvents(t, m, eventlog.Filter{})) + vm.Memory.Usage, vm.Memory.Used, vm.Memory.Free = 10, 100, 900 + vm.Memory.Observation.State = "last-known" + vm.Disk.Usage, vm.DiskStatusReason = 10, reason + for i := range vm.Disks { + vm.Disks[i].Usage = 10 + } + m.CheckGuest(vm, "site") + vm.Disks = nil // Failed/expired retained inventory is not removal. + vm.Disk.Usage = -1 + for poll := 0; poll < 4; poll++ { + m.CheckGuest(vm, "site") + } + if after := guestObservationBytes(t, m); !bytes.Equal(before, after) { + t.Fatalf("unknown observation changed incident values/identity/ack/age: before=%s after=%s", before, after) + } + if len(queryAlertEvents(t, m, eventlog.Filter{})) != events { + t.Fatal("unknown observation emitted lifecycle/delivery evidence") + } + vm.Memory.Observation.State = "current" + vm.Disk.Usage, vm.DiskStatusReason = 10, "" + vm.Disks = []models.Disk{{Total: 1000, Used: 100, Free: 900, Usage: 10, Mountpoint: "/", Device: "/dev/vda"}} + m.CheckGuest(vm, "site") // Fresh recovery plus authoritative removal of /data. + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 0 || len(queryAlertEvents(t, m, eventlog.Filter{Types: []string{eventlog.TypeResolved}})) != 3 { + t.Fatal("fresh recovery/removal must resolve each original incident exactly once") + } + }) + } +} + +func TestGuestAlertRetainedAggregateAndPendingCannotAdvance(t *testing.T) { + m := guestObservationManager(t) + vm := guestObservationVM() + vm.Disks = nil + vm.Memory.Usage = 90 + m.mu.Lock() + m.config.MetricTimeThresholds = map[string]map[string]int{"guest": {"memory": 60, "disk": 60}} + m.mu.Unlock() + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 0 { + t.Fatal("fresh warning observations should begin their delay, not fire") + } + m.mu.Lock() + m.core.ShiftPending(-2 * time.Minute) + m.mu.Unlock() + vm.Memory.Observation.State, vm.DiskStatusReason = "last-known", "prev-agent-cooldown" + for poll := 0; poll < 4; poll++ { + m.CheckGuest(vm, "site") + } + if len(m.GetActiveAlerts()) != 0 { + t.Fatal("retained memory/disk promoted a delayed observation to an incident") + } + vm.Memory.Observation.State, vm.DiskStatusReason = "current", "" + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 2 { + t.Fatal("fresh evidence must resume the existing delay policy") + } +} + +func TestGuestAlertFreshIndependentMetricsAndDisablement(t *testing.T) { + for _, state := range []string{"", "current"} { + t.Run("legacy-or-current/"+state, func(t *testing.T) { + m := guestObservationManager(t) + vm := guestObservationVM() + vm.Memory.Observation.State = state + // Guest-agent state alone cannot suppress independent current data. + vm.GuestAgentStatus, vm.Lock = "deferred", "backup" + vm.Memory.Observation.Source = "agent" + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 3 { + t.Fatal("legacy/current independent memory and filesystems were suppressed") + } + vm.Memory.Observation.State, vm.DiskStatusReason = "last-known", "prev-vm-locked" + cfg := m.GetConfig() + cfg.GuestDefaults.Memory = &HysteresisThreshold{Trigger: -1} + cfg.GuestDefaults.Disk = &HysteresisThreshold{Trigger: -1} + m.UpdateConfig(cfg) + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 0 { + t.Fatal("explicit rule disablement must remain independent of telemetry") + } + }) + } + m := guestObservationManager(t) + vm := guestObservationVM() + vm.Memory.Observation.State, vm.DiskStatusReason = "last-known", "prev-vm-locked" + vm.CPU, vm.DiskWrite = .9, 10*1024*1024 + vm.IORateValidity.DiskWrite = true + m.CheckGuest(vm, "site") + active := m.GetActiveAlerts() + if len(active) != 2 { + t.Fatalf("current CPU/I/O should still fire, retained memory/disk should not: %+v", active) + } + for _, a := range active { + if a.Type != "cpu" && a.Type != "diskWrite" { + t.Fatalf("unexpected retained metric alert: %+v", a) + } + } +} + +func TestGuestFiltersDoNotUseRetainedObservations(t *testing.T) { + m := guestObservationManager(t) + vm := guestObservationVM() + for _, field := range []string{"memory", "disk"} { + for _, op := range []string{">", "<", "=="} { + condition := FilterCondition{Type: "metric", Field: field, Operator: op, Value: 96.0} + vm.Memory.Observation.State, vm.DiskStatusReason = "last-known", "prev-vm-locked" + if m.evaluateVMCondition(vm, condition) || m.evaluateFilterCondition(&vm, condition) { + t.Fatalf("retained %s matched %s", field, op) + } + } + } + vm.Memory.Observation.State, vm.DiskStatusReason = "current", "" + for _, field := range []string{"memory", "disk"} { + if !m.evaluateVMCondition(vm, FilterCondition{Type: "metric", Field: field, Operator: ">", Value: 90.0}) { + t.Fatalf("fresh %s could not match", field) + } + } + vm.Memory = models.UnavailableMemory(1000) + if m.evaluateVMCondition(vm, FilterCondition{Type: "metric", Field: "memory", Operator: "<", Value: 90.0}) { + t.Fatal("unavailable memory was treated as a measured zero in a filter") + } +} diff --git a/internal/alerts/guest_snapshot.go b/internal/alerts/guest_snapshot.go index f9741f9647..9cbcc805e0 100644 --- a/internal/alerts/guest_snapshot.go +++ b/internal/alerts/guest_snapshot.go @@ -28,9 +28,10 @@ type guestSnapshot struct { CPUPercent float64 MemUsage float64 - // MemoryUnavailable prevents a missing cache-aware sample from clearing a - // real alert or starting a false one. + // Unavailable observations include retained display values. They cannot + // supply a new breach, recovery, filter match or filesystem removal. MemoryUnavailable bool + DiskUnavailable bool DiskUsage float64 DiskRead int64 DiskWrite int64 @@ -81,18 +82,20 @@ func (g guestSnapshot) resourceType() string { func (g guestSnapshot) metrics() guestMetrics { return guestMetrics{ - CPU: g.CPUPercent, - MemUsage: g.MemUsage, - DiskUsage: g.DiskUsage, - DiskRead: g.DiskRead, - DiskWrite: g.DiskWrite, - NetworkIn: g.NetworkIn, - NetworkOut: g.NetworkOut, - IORateValidity: g.IORateValidity, - Name: g.Name, - Node: g.Node, - ID: g.ID, - Status: g.Status, + CPU: g.CPUPercent, + MemUsage: g.MemUsage, + DiskUsage: g.DiskUsage, + MemoryUnavailable: g.MemoryUnavailable, + DiskUnavailable: g.DiskUnavailable, + DiskRead: g.DiskRead, + DiskWrite: g.DiskWrite, + NetworkIn: g.NetworkIn, + NetworkOut: g.NetworkOut, + IORateValidity: g.IORateValidity, + Name: g.Name, + Node: g.Node, + ID: g.ID, + Status: g.Status, } } @@ -108,7 +111,8 @@ func guestSnapshotFromVM(vm models.VM) guestSnapshot { Lock: vm.Lock, CPUPercent: unifiedresources.ProxmoxGuestCPUPercent(vm.CPU), MemUsage: vm.Memory.Usage, - MemoryUnavailable: vm.Memory.UsageUnavailable, + MemoryUnavailable: guestMemoryUnavailableForAlerts(vm.Memory), + DiskUnavailable: vm.DiskStatusReason != "" || vm.Disk.Usage < 0, DiskUsage: vm.Disk.Usage, DiskRead: vm.DiskRead, DiskWrite: vm.DiskWrite, @@ -133,7 +137,8 @@ func guestSnapshotFromContainer(container models.Container) guestSnapshot { Lock: container.Lock, CPUPercent: unifiedresources.ProxmoxGuestCPUPercent(container.CPU), MemUsage: container.Memory.Usage, - MemoryUnavailable: container.Memory.UsageUnavailable, + MemoryUnavailable: guestMemoryUnavailableForAlerts(container.Memory), + DiskUnavailable: container.Disk.Usage < 0, DiskUsage: container.Disk.Usage, DiskRead: container.DiskRead, DiskWrite: container.DiskWrite, @@ -146,6 +151,13 @@ func guestSnapshotFromContainer(container models.Container) guestSnapshot { }.normalizeCollections() } +func guestMemoryUnavailableForAlerts(memory models.Memory) bool { + // The producer owns freshness and source selection. Legacy values without + // an observation retain their existing behavior; do not infer freshness + // from disk/QGA state, since PVE or a linked Pulse agent can be independent. + return memory.UsageUnavailable || (memory.Observation.State != "" && memory.Observation.State != "current") +} + func guestKindFromType(guestType string) guestKind { switch strings.ToLower(strings.TrimSpace(guestType)) { case "qemu", "vm": diff --git a/internal/alerts/guest_snapshot_test.go b/internal/alerts/guest_snapshot_test.go index 28c7e4e65f..2f89b24eda 100644 --- a/internal/alerts/guest_snapshot_test.go +++ b/internal/alerts/guest_snapshot_test.go @@ -19,6 +19,72 @@ func TestGuestSnapshotResourceTypeUsesCanonicalSystemContainer(t *testing.T) { } } +func TestGuestSnapshotObservationAvailability(t *testing.T) { + for _, test := range []struct { + name string + memory models.Memory + disk models.Disk + reason string + wantMemory bool + wantDisk bool + }{ + {name: "legacy measured zero", memory: models.Memory{}, disk: models.Disk{}}, + {name: "current independent read", memory: models.Memory{Usage: 96, Observation: models.MemoryObservation{State: "current", Source: "agent"}}, disk: models.Disk{Usage: 98}}, + {name: "retained memory only", memory: models.Memory{Usage: 96, Observation: models.MemoryObservation{State: "last-known"}}, disk: models.Disk{Usage: 98}, wantMemory: true}, + {name: "unavailable numeric placeholder", memory: models.Memory{Usage: 0, Observation: models.MemoryObservation{State: "unavailable"}}, wantMemory: true}, + {name: "unknown observation version", memory: models.Memory{Usage: 96, Observation: models.MemoryObservation{State: "future-version"}}, wantMemory: true}, + {name: "explicit unavailable overrides current", memory: models.Memory{Usage: 96, UsageUnavailable: true, Observation: models.MemoryObservation{State: "current"}}, wantMemory: true}, + {name: "retained filesystem only", memory: models.Memory{Usage: 96}, disk: models.Disk{Usage: 98}, reason: "prev-vm-locked", wantDisk: true}, + {name: "failed empty filesystem inventory", memory: models.Memory{Usage: 96}, disk: models.Disk{Usage: -1}, reason: "agent-error", wantDisk: true}, + {name: "legacy unavailable aggregate", memory: models.Memory{Usage: 96}, disk: models.Disk{Usage: -1}, wantDisk: true}, + } { + t.Run(test.name, func(t *testing.T) { + // A deferred QGA session does not by itself make independent + // selected memory or Pulse-agent filesystems unavailable. + vm := models.VM{Memory: test.memory, Disk: test.disk, DiskStatusReason: test.reason, GuestAgentStatus: "deferred"} + for _, guest := range []any{vm, &vm} { + snapshot, ok := extractGuestSnapshot(guest) + if !ok || snapshot.MemoryUnavailable != test.wantMemory || snapshot.DiskUnavailable != test.wantDisk { + t.Fatalf("observation flags lost: %+v", snapshot) + } + metrics := snapshot.metrics() + if metrics.MemoryUnavailable != test.wantMemory || metrics.DiskUnavailable != test.wantDisk || metrics.MemUsage != test.memory.Usage || metrics.DiskUsage != test.disk.Usage { + t.Fatal("filter projection lost availability or rewrote retained values") + } + } + ct := models.Container{Memory: test.memory, Disk: test.disk} + for _, guest := range []any{ct, &ct} { + snapshot, _ := extractGuestSnapshot(guest) + if snapshot.MemoryUnavailable != test.wantMemory || snapshot.DiskUnavailable != (test.disk.Usage < 0) { + t.Fatal("LXC observation availability changed with an unrelated VM disk failure") + } + } + }) + } +} + +func TestGuestAlertRetainedDiskStillHonoursExplicitOverride(t *testing.T) { + for _, override := range []ThresholdConfig{{Disabled: true}, {Disk: &HysteresisThreshold{Trigger: -1}}} { + m := guestObservationManager(t) + vm := guestObservationVM() + vm.Memory.Usage = 10 + m.CheckGuest(vm, "site") + if len(m.GetActiveAlerts()) != 2 { + t.Fatal("expected two current filesystem incidents") + } + _, keySource, key := guestDiskIdentity(vm.Disks[0], 0) + cfg := m.GetConfig() + cfg.Overrides[guestDiskOverrideKey(vm, vm.ID, keySource)] = override + m.UpdateConfig(cfg) + vm.DiskStatusReason = "prev-vm-locked" + m.CheckGuest(vm, "site") + active := m.GetActiveAlerts() + if len(active) != 1 || active[0].ResourceID == vm.ID+"-disk-"+key { + t.Fatalf("explicit filesystem disablement was lost or crossed disk identity: %+v", active) + } + } +} + func TestGuestSnapshotUsesCanonicalProxmoxCPUPercent(t *testing.T) { for _, tc := range []struct { name string diff --git a/internal/monitoring/guest_alert_observation_integration_test.go b/internal/monitoring/guest_alert_observation_integration_test.go new file mode 100644 index 0000000000..ec6c293914 --- /dev/null +++ b/internal/monitoring/guest_alert_observation_integration_test.go @@ -0,0 +1,140 @@ +package monitoring + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "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" +) + +// Exercise actual HTTP client, builders, saved snapshots, unified views, History +// and the production alert call. No native QGA or backup is performed. +func TestGuestBackupDeferralAlertObservationLifecycle(t *testing.T) { + const mib = uint64(1024 * 1024) + var phase, guestCalls atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasSuffix(r.URL.Path, "/config"): + if phase.Load() == 1 { + fmt.Fprint(w, `{"data":{"lock":"backup"}}`) + } 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}}`, 1000*mib, 1000*mib) + case strings.Contains(r.URL.Path, "/agent/"): + guestCalls.Add(1) + switch { + case strings.HasSuffix(r.URL.Path, "file-read"): + available := 40 * 1024 + if phase.Load() == 2 { + available = 800 * 1024 + } + json.NewEncoder(w).Encode(map[string]any{"data": map[string]any{"content": fmt.Sprintf("MemTotal: 1024000 kB\nMemFree: 10240 kB\nMemAvailable: %d kB\n", available)}}) + case strings.HasSuffix(r.URL.Path, "get-fsinfo"): + used := 980 * mib + if phase.Load() == 2 { + used = 200 * mib + } + fmt.Fprintf(w, `{"data":{"result":[{"mountpoint":"/","type":"ext4","total-bytes":%d,"used-bytes":%d}]}}`, 1000*mib, used) + case strings.HasSuffix(r.URL.Path, "network-get-interfaces"): + fmt.Fprint(w, `{"data":{"result":[]}}`) + case strings.HasSuffix(r.URL.Path, "get-osinfo"): + fmt.Fprint(w, `{"data":{"name":"Linux","version":"fixture"}}`) + default: + fmt.Fprint(w, `{"data":{"result":{"version":"1.0"}}}`) + } + 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) + } + am := alerts.NewManagerWithDataDir(t.TempDir(), alerts.WithoutPersistedAlertRestore()) + defer am.Stop() + cfg := am.GetConfig() + cfg.Enabled = true + cfg.GuestDefaults.Memory = &alerts.HysteresisThreshold{Trigger: 85, Clear: 80} + cfg.GuestDefaults.Disk = &alerts.HysteresisThreshold{Trigger: 90, Clear: 85} + cfg.GuestDefaults.CPU = &alerts.HysteresisThreshold{Trigger: 80, Clear: 75} + cfg.MetricTimeThresholds = map[string]map[string]int{"guest": {"memory": 0, "disk": 0, "cpu": 0}} + am.UpdateConfig(cfg) + registry := unifiedresources.NewRegistry(nil) + m := &Monitor{config: &config.Config{}, alertManager: am, rateTracker: NewRateTracker(), metricsHistory: NewMetricsHistory(32, time.Hour), guestMetadataLimiter: make(map[string]time.Time)} + res := proxmox.ClusterResource{Type: "qemu", Node: "node", Name: "guest", VMID: 105, Status: "running", MaxMem: 1000 * mib, Mem: 1000 * mib, MaxDisk: 1000 * mib} + identity := makeGuestID("alert-observation", "node", 105) + var previous *models.VM + build := func() models.VM { + t.Helper() + vm, raw, source, notes, at, ok := m.buildVMFromClusterResource(context.Background(), "alert-observation", res, client, identity, nil, previous) + if !ok { + t.Fatal("VM disappeared") + } + m.recordGuestSnapshot("alert-observation", "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)) + registry.IngestSnapshot(models.StateSnapshot{VMs: []models.VM{vm}}) + next := previousVMFromView(registry.VMs()[0]) + previous = &next + return vm + } + initial := build() + // The observation was made before alert policy became enabled for this VM. + // A retained high value must not create an incident upon the next poll. + phase.Store(1) + before := guestCalls.Load() + locked := build() + if locked.Memory.Observation.State != "last-known" || !strings.HasPrefix(locked.DiskStatusReason, "prev-") { + t.Fatalf("fixture did not produce real last-known evidence: %+v", locked) + } + m.checkGuestAlertsForVM("alert-observation", locked) + if len(am.GetActiveAlerts()) != 0 { + t.Fatal("production alert call treated retained backup-lock evidence as current") + } + // The original current evidence can establish an alert. Subsequent lock + // polls must neither refresh its observation nor resolve it on expiry. + m.checkGuestAlertsForVM("alert-observation", initial) + if len(am.GetActiveAlerts()) != 2 { + t.Fatal("current memory/filesystem evidence did not establish two alerts") + } + for poll := 0; poll < 4; poll++ { + m.checkGuestAlertsForVM("alert-observation", build()) + } + if guestCalls.Load() != before || len(am.GetActiveAlerts()) != 2 { + t.Fatal("backup deferral queried QGA or lost the original alerts") + } + for _, a := range am.GetActiveAlerts() { + if a.Value != 96 && a.Value != 98 { + t.Fatal("retained poll changed the last trusted alert value") + } + } + if got := len(m.metricsHistory.GetGuestMetrics(identity, "memory", time.Hour)); got != 1 { + t.Fatalf("retained memory refreshed History: %d points", got) + } + if got := len(m.metricsHistory.GetGuestMetrics(identity, "disk", time.Hour)); got != 1 { + t.Fatalf("retained filesystem refreshed History: %d points", got) + } + phase.Store(2) + entry := m.vmAgentMemCache[guestMemoryCacheKey("alert-observation", "node", 105)] + entry.fetchedAt = time.Now().Add(-2 * vmAgentMemCacheTTL) + m.vmAgentMemCache[guestMemoryCacheKey("alert-observation", "node", 105)] = entry + resumed := build() + m.checkGuestAlertsForVM("alert-observation", resumed) + if guestCalls.Load() <= before || resumed.Memory.Observation.State != "current" || resumed.DiskStatusReason != "" || len(am.GetActiveAlerts()) != 0 { + t.Fatal("fresh unlocked memory/filesystem evidence did not recover the original incidents") + } +}