mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
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
This commit is contained in:
pulse-triage[bot]
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user