From b0f8b96c8a159ac242b85f2d09f8dfffd5c29c7a Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 4 Oct 2026 21:37:11 +0100 Subject: [PATCH] Reproduce retained guest readings changing alert state Exercise production memory/filesystem alert admission, delayed breaches, filters, canonical recovery and the connected HTTP backup-lock polling boundary. Retained evidence is not a new observation or an authoritative filesystem removal. Change-source: pulse-maintainer --- internal/alerts/guest_observation_test.go | 220 ++++++++++++++++++ ...uest_alert_observation_integration_test.go | 140 +++++++++++ 2 files changed, 360 insertions(+) create mode 100644 internal/alerts/guest_observation_test.go create mode 100644 internal/monitoring/guest_alert_observation_integration_test.go 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/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") + } +}