diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index b1ce10787..e98ded7b0 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -618,6 +618,14 @@ installer download and the agent's subsequent Pulse TLS connection. ## Shared Boundaries +The shared `PBSInstance.NodeMetricsUnavailable` field belongs exclusively to +provider polling and alert evaluation. It is retained by in-process state copies +but excluded from JSON; it neither grants nor revokes host-agent identity, +enrolment, removal or command authority. No agent lifecycle decision may infer +host removal or re-enrolment from a PBS node-status failure. The shared-model +boundary is verified in `internal/models/metrics_types_test.go`; this is not +new host-agent removal or installed re-enrolment qualification. + `internal/models/models.go` and `internal/monitoring/monitor.go` also carry monitoring-owned Proxmox cluster node identity and membership-confirmation state. Provider instances with equal cluster/member display names must not diff --git a/docs/release-control/v6/internal/subsystems/alerts.md b/docs/release-control/v6/internal/subsystems/alerts.md index 14d3e2ae5..d2ea7c64f 100644 --- a/docs/release-control/v6/internal/subsystems/alerts.md +++ b/docs/release-control/v6/internal/subsystems/alerts.md @@ -373,6 +373,15 @@ default construction path still restores. ## Shared Boundaries +PBS node-status availability is distinct from connectivity. While a connected +PBS carries monitoring-owned `NodeMetricsUnavailable` evidence, CPU and memory +evaluation must not treat zero-valued placeholders as recovery. Existing +policy suppression and full-outage handling retain precedence. Valid low +measurements resume normal recovery, including callbacks and recent history. +Regression proof lives in `internal/alerts/telemetry_quality_test.go` and the +HTTP-to-manager lifecycle in `internal/monitoring/monitor_pbs_coverage_test.go`. + + 1. `frontend-modern/src/stores/websocket.ts` shared with `performance-and-scalability`: the connection-owned realtime store is both the canonical alert truth boundary and the fleet-scale resource reconciliation hot path. That shared store normalizes slimmed broadcast resources at ingestion — expanding `capabilitiesRef` through the state `capabilityCatalog` and diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index c2213b24a..4956c257b 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -17,6 +17,39 @@ ## Purpose +PBS node-status collection must reject HTTP-success responses whose `data` +is omitted or null (including a null response envelope). Absent status is +unavailable telemetry, not measured zero usage: the poller retains independently +established connectivity, marks node metrics unavailable, and must not resolve +an active metric incident from that response. This does not add per-field +validation of populated status objects. + +Verification: `TestClient_GetNodeStatus_MissingData` in +`pkg/pbs/client_http_test.go` covers the absent envelopes. +`TestPBSPartialMetricsWebhookLifecycle` in +`internal/monitoring/monitor_pbs_webhook_test.go` exercises repeated null status +through the real poller, alert manager, notification queue and local webhook. +It requires online-but-unavailable projection, unchanged incident identity, +no false recovery history or delivery, then one identity-preserving recovery +after valid low-memory samples. These are local synthetic checks, not +installed-artifact or off-host destination qualification. + +Direct PBS backup polling classifies typed API and authentication failures by +the response status exposed by `pbs.HTTPStatus`, including wrapped errors. +A 5xx gateway or server response remains transient even when its body quotes +an upstream “API error 403” or “API error 404”; that text must not erase the +last known backup inventory. Genuine 4xx responses remain terminal under the +existing cache policy. The legacy untyped-error fallback is unchanged. +Retaining cached inventory does not establish a successful poll or fresh +backup evidence. + +Verification: `TestPollPBSBackups_PreservesCacheOnTransientDatastoreError` and +`TestPollPBSBackups_DropsStaleCacheOnTerminalDatastoreError` in +`internal/monitoring/monitor_backups_readstate_test.go` exercise actual HTTP +fixtures for 500, 502 quoting 403, 503 quoting 404, and genuine 401/403/404. +These tests prove cache retention/removal, not installed PBS wake, service +restart, or notification receipt. + TrueNAS CRITICAL, ALERT and EMERGENCY native levels project as canonical critical incidents. EMERGENCY must not be discarded as an unknown level: repeated observations retain the active incident rather than supplying false recovery evidence. Provider projection tests cover every documented native level and normalized input. TrueNAS native alert projection preserves the trimmed, uppercase provider level in ResourceIncident.NativeSeverity. INFO and NOTICE retain the same canonical monitor risk; consumers must not lose their distinct actionability when projecting provider evidence. @@ -653,6 +686,17 @@ cleanup so readers cannot retain orphaned runtime or alert projections. ## Shared Boundaries +PBS polling owns the internal `PBSInstance.NodeMetricsUnavailable` discriminator: +each poll starts unavailable and only a successful non-nil node-status result +clears it. A denied or failed node-status endpoint does not invalidate successful +connectivity or independently accessible datastore inventory. In-process state +copies preserve this evidence; JSON deliberately does not carry it. The +zero-value compatibility default is not persisted availability evidence. +Proof: `internal/models/metrics_types_test.go` and +`internal/monitoring/monitor_pbs_coverage_test.go`. The latter exercises real +HTTP polling through normal alert-manager publication, not destination delivery. + + 1. `internal/config/host_continuity.go` shared with `agent-lifecycle`: the durable host identity, report-order watermark, and removal tombstone journal is jointly owned by agent lifecycle admission and monitoring report continuity. 2. `internal/kubernetesagent/agent.go` shared with `agent-lifecycle`: the Kubernetes native agent runtime is both a monitoring inventory source and an agent lifecycle Pulse control-plane transport client. 3. `internal/mock/fixture_graph.go` shared with `performance-and-scalability`: the canonical mock fixture graph is both monitoring-owned runtime data and a protected large-estate demo transport hot path. diff --git a/frontend-modern/src/__tests__/useAppRuntimeState.test.ts b/frontend-modern/src/__tests__/useAppRuntimeState.test.ts index 7b65aee58..908220226 100644 --- a/frontend-modern/src/__tests__/useAppRuntimeState.test.ts +++ b/frontend-modern/src/__tests__/useAppRuntimeState.test.ts @@ -77,6 +77,7 @@ describe('useAppRuntimeState', () => { let websocketConnected: boolean; let websocketReconnecting: boolean; let websocketInitialDataReceived: boolean; + let websocketResourceSnapshotReceived: boolean; beforeEach(async () => { vi.resetModules(); @@ -148,6 +149,7 @@ describe('useAppRuntimeState', () => { websocketConnected = false; websocketReconnecting = false; websocketInitialDataReceived = false; + websocketResourceSnapshotReceived = false; vi.doMock('@/stores/websocket-global', () => ({ getGlobalWebSocketStore: () => ({ @@ -155,7 +157,7 @@ describe('useAppRuntimeState', () => { connected: () => websocketConnected, reconnecting: () => websocketReconnecting, initialDataReceived: () => websocketInitialDataReceived, - resourceSnapshotReceived: () => websocketInitialDataReceived, + resourceSnapshotReceived: () => websocketResourceSnapshotReceived, reconnect: vi.fn(), switchUrl: vi.fn(), }), @@ -700,7 +702,8 @@ describe('useAppRuntimeState', () => { websocketState = makeWebSocketState({ activeAlerts: [{ id: 'recovered-alert' } as State['activeAlerts'][number]] }); websocketConnected = false; websocketReconnecting = true; - websocketInitialDataReceived = false; + websocketInitialDataReceived = true; + websocketResourceSnapshotReceived = false; const { hookState, dispose } = mountHook(); await waitFor(() => expect(hookState.enhancedStore()).not.toBeNull()); expect(hookState.state().activeAlerts).toHaveLength(1); @@ -708,6 +711,20 @@ describe('useAppRuntimeState', () => { dispose(); }); + it('retains explicit empty resource admission while transport hydration resets', async () => { + websocketState = makeWebSocketState(); + websocketConnected = false; + websocketReconnecting = true; + websocketInitialDataReceived = false; + websocketResourceSnapshotReceived = true; + const { hookState, dispose } = mountHook(); + await waitFor(() => expect(hookState.enhancedStore()).not.toBeNull()); + expect(hookState.state().resources).toHaveLength(0); + expect(hookState.enhancedStore()?.initialDataReceived()).toBe(false); + expect(hookState.runtimeStateResolved()).toBe(true); + dispose(); + }); + it('distinguishes an evidence-free first load from an authenticated empty estate', async () => { // The distinction still matters: an estate with nothing in it must resolve // to "no platform pages", not sit unresolved forever. It is now answered by diff --git a/internal/alerts/pbs.go b/internal/alerts/pbs.go index c07a0ca65..b7575126e 100644 --- a/internal/alerts/pbs.go +++ b/internal/alerts/pbs.go @@ -125,6 +125,12 @@ func (m *Manager) CheckPBS(pbs models.PBSInstance) { return } + // Endpoint failure is not evidence of metric recovery. Connectivity above + // remains independent, as do explicit policy suppression and full outages. + if pbs.NodeMetricsUnavailable { + return + } + m.evaluateUnifiedMetrics(&UnifiedResourceInput{ ID: pbs.ID, Type: "pbs", diff --git a/internal/alerts/pbs_restart_test.go b/internal/alerts/pbs_restart_test.go new file mode 100644 index 000000000..d283d4769 --- /dev/null +++ b/internal/alerts/pbs_restart_test.go @@ -0,0 +1,87 @@ +package alerts + +import ( + "testing" + + "github.com/rcourtman/pulse-go-rewrite/internal/alerts/eventlog" + "github.com/rcourtman/pulse-go-rewrite/internal/models" +) + +// Exercise real threshold transitions across disk-backed manager lifetimes, +// rather than restoring manually seeded active-alert fixtures. +func TestPBSMissingMetricsAcrossRestart(t *testing.T) { + dataDir := t.TempDir() + start := func() *Manager { + m := NewManagerWithDataDir(dataDir) + t.Cleanup(m.Stop) + m.EnableEventLog() + if !m.activeStateAuthoritative.Load() { + t.Fatal("SQLite active state is not authoritative") + } + m.UpdateConfig(AlertConfig{Enabled: true, PBSDefaults: ThresholdConfig{ + CPU: &HysteresisThreshold{Trigger: 80, Clear: 75}, + Memory: &HysteresisThreshold{Trigger: 85, Clear: 80}, + }}) + disableTestTimeThresholds(m) + return m + } + assertEvents := func(m *Manager, fired, resolved int) { + t.Helper() + for kind, want := range map[string]int{eventlog.TypeFired: fired, eventlog.TypeResolved: resolved} { + if got := len(queryAlertEvents(t, m, eventlog.Filter{Types: []string{kind}})); got != want { + t.Fatalf("%s events = %d, want %d", kind, got, want) + } + } + } + + m := start() + p := models.PBSInstance{ID: "pbs-restart", Name: "backup", Status: "online", CPU: 95, Memory: 95} + m.CheckPBS(p) + initial := m.GetActiveAlerts() + if len(initial) != 2 { + t.Fatalf("initial incidents = %d, want 2", len(initial)) + } + assertEvents(m, 2, 0) + m.Stop() + + m = start() + p.CPU, p.Memory, p.NodeMetricsUnavailable = 0, 0, true + for range 5 { + m.CheckPBS(p) + } + restored := m.GetActiveAlerts() + if len(restored) != len(initial) { + t.Fatalf("missing metrics after restart retained %d incidents, want 2", len(restored)) + } + for _, before := range initial { + found := false + for _, after := range restored { + if before.ID == after.ID && before.StartTime.Equal(after.StartTime) { + found = true + } + } + if !found { + t.Fatalf("incident identity/start time changed across restart: %s", before.ID) + } + } + assertEvents(m, 2, 0) + + p.NodeMetricsUnavailable = false + for range 5 { + m.CheckPBS(p) + } + if got := len(m.GetActiveAlerts()); got != 0 { + t.Fatalf("measured zero retained %d incidents", got) + } + assertEvents(m, 2, 2) + m.Stop() + + m = start() + if got := len(m.GetActiveAlerts()); got != 0 { + t.Fatalf("second restart resurrected %d resolved incidents", got) + } + assertEvents(m, 2, 2) + if got := len(m.GetAlertHistory(10)); got != 2 { + t.Fatalf("durable incident history = %d, want 2", got) + } +} diff --git a/internal/alerts/telemetry_quality_test.go b/internal/alerts/telemetry_quality_test.go index e82d90e08..4a336b53d 100644 --- a/internal/alerts/telemetry_quality_test.go +++ b/internal/alerts/telemetry_quality_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" ) @@ -192,3 +193,46 @@ func TestAlertQualityResolutionDurationBoundaries(t *testing.T) { } func alertQualityTimePtr(value time.Time) *time.Time { return &value } + +func TestPBSMissingMetricsDoNotResolve(t *testing.T) { + m := newUnifiedEvalParityManager(t) + m.UpdateConfig(AlertConfig{Enabled: true, PBSDefaults: ThresholdConfig{ + CPU: &HysteresisThreshold{Trigger: 80, Clear: 75}, + Memory: &HysteresisThreshold{Trigger: 85, Clear: 80}, + }}) + disableTestTimeThresholds(m) + resolved := make(chan string, 8) + m.SetResolvedCallback(func(id string) { resolved <- id }) + p := models.PBSInstance{ID: "pbs-missing", Name: "backup", Status: "online", CPU: 95, Memory: 95} + m.CheckPBS(p) + if len(m.GetActiveAlerts()) != 2 { + t.Fatal("expected two high utilisation alerts") + } + p.CPU, p.Memory = 0, 0 + p.NodeMetricsUnavailable = true + for range 5 { + m.CheckPBS(p) + } + if len(m.GetActiveAlerts()) != 2 { + t.Fatal("missing metrics falsely resolved high utilisation") + } + select { + case id := <-resolved: + t.Fatalf("false recovery callback %s", id) + case <-time.After(50 * time.Millisecond): + } + p.NodeMetricsUnavailable = false + for range 5 { + m.CheckPBS(p) + } + if len(m.GetActiveAlerts()) != 0 { + t.Fatal("measured zero did not recover") + } + for range 2 { + select { + case <-resolved: + case <-time.After(time.Second): + t.Fatal("missing genuine recovery callback") + } + } +} diff --git a/internal/models/metrics_types_test.go b/internal/models/metrics_types_test.go index 1b839bc69..2b53b9354 100644 --- a/internal/models/metrics_types_test.go +++ b/internal/models/metrics_types_test.go @@ -343,3 +343,40 @@ func TestPBSGuestConfirmationEvidenceStaysOutOfSerializedState(t *testing.T) { t.Fatal("PBS guest confirmation evidence must not appear in snapshots") } } + +// PBS poll evidence survives in-process copying but is not an agent wire input. +func TestPBSNodeMetricAvailabilityModelBoundary(t *testing.T) { + state := NewState() + state.UpdatePBSInstance(PBSInstance{ + ID: "pbs-availability", Name: "pbs", Status: "online", + NodeMetricsUnavailable: true, + }) + snapshot := state.GetSnapshot() + if len(snapshot.PBSInstances) != 1 || !snapshot.PBSInstances[0].NodeMetricsUnavailable { + t.Fatal("snapshot lost unavailable-node-metrics evidence") + } + snapshot.PBSInstances[0].NodeMetricsUnavailable = false + if !state.GetSnapshot().PBSInstances[0].NodeMetricsUnavailable { + t.Fatal("snapshot mutation changed authoritative poll evidence") + } + missing := state.GetSnapshot().PBSInstances[0] + wire, err := json.Marshal(missing) + if err != nil { + t.Fatal(err) + } + missing.NodeMetricsUnavailable = false + availableWire, err := json.Marshal(missing) + if err != nil { + t.Fatal(err) + } + if string(wire) != string(availableWire) { + t.Fatal("internal availability evidence changed the public wire contract") + } + var decoded PBSInstance + if err := json.Unmarshal([]byte(`{"id":"pbs","NodeMetricsUnavailable":true,"nodeMetricsUnavailable":true}`), &decoded); err != nil { + t.Fatal(err) + } + if decoded.NodeMetricsUnavailable { + t.Fatal("wire input asserted monitoring-owned availability evidence") + } +} diff --git a/internal/models/models.go b/internal/models/models.go index 318c8631a..f0e979774 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -2699,6 +2699,10 @@ type PhysicalDisk struct { // PBSInstance represents a Proxmox Backup Server instance type PBSInstance struct { + // NodeMetricsUnavailable distinguishes a failed node-status poll from measured zero. + // Internal poll evidence; the zero value preserves legacy in-process callers. + NodeMetricsUnavailable bool `json:"-"` + ID string `json:"id"` Name string `json:"name"` Host string `json:"host"` diff --git a/internal/monitoring/backup_guard.go b/internal/monitoring/backup_guard.go index 388768b02..f6e1b298d 100644 --- a/internal/monitoring/backup_guard.go +++ b/internal/monitoring/backup_guard.go @@ -1,6 +1,10 @@ package monitoring -import "strings" +import ( + "strings" + + "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" +) // shouldPreservePBSBackupsWithTerminal preserves stale PBS backups only when all // datastore fetches failed and at least one failure was non-terminal. @@ -17,6 +21,10 @@ func shouldReuseCachedPBSBackups(err error) bool { if err == nil { return false } + if status, ok := pbs.HTTPStatus(err); ok { + return status < 400 || status >= 500 + } + // Retain compatibility with untyped errors from older callers. if strings.Contains(strings.ToLower(err.Error()), "api error 4") { return false } diff --git a/internal/monitoring/monitor_backups_readstate_test.go b/internal/monitoring/monitor_backups_readstate_test.go index 1baf09269..bd2b881d6 100644 --- a/internal/monitoring/monitor_backups_readstate_test.go +++ b/internal/monitoring/monitor_backups_readstate_test.go @@ -995,3 +995,117 @@ func TestRetirePVEInstanceRuntimeClearsPBSGuestConfirmations(t *testing.T) { } } } + +func TestPollPBSBackups_DropsStaleCacheOnTerminalDatastoreError(t *testing.T) { + t.Parallel() + + for _, status := range []int{http.StatusUnauthorized, http.StatusForbidden, http.StatusNotFound} { + t.Run(http.StatusText(status), func(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/admin/datastore/archive/groups") { + http.Error(w, `{"errors":"datastore does not exist"}`, status) + return + } + http.NotFound(w, r) + })) + defer server.Close() + + client, err := pbs.NewClient(pbs.ClientConfig{ + Host: server.URL, + TokenName: "root@pam!token", + TokenValue: "secret", + }) + if err != nil { + t.Fatalf("failed to create PBS client: %v", err) + } + + m := &Monitor{state: models.NewState()} + m.state.UpdatePBSBackups("pbs1", []models.PBSBackup{ + { + ID: "pbs-pbs1-archive--vm-100-1700000000", + Instance: "pbs1", + Datastore: "archive", + Namespace: "", + BackupType: "vm", + VMID: "100", + BackupTime: time.Unix(1700000000, 0), + }, + }) + + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ + {Name: "archive"}, + }) + + snapshot := m.state.GetSnapshot() + for _, backup := range snapshot.PBSBackups { + if backup.Instance == "pbs1" { + t.Fatalf("expected stale backups to be removed after terminal error, found: %+v", backup) + } + } + }) + } +} + +func TestPollPBSBackups_PreservesCacheOnTransientDatastoreError(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + status int + body string + }{ + {"server error", http.StatusInternalServerError, "temporary server issue"}, + {"gateway quoting forbidden", http.StatusBadGateway, "upstream API error 403: permission denied"}, + {"unavailable quoting missing datastore", http.StatusServiceUnavailable, "upstream API error 404: datastore does not exist"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/admin/datastore/archive/groups") { + http.Error(w, tc.body, tc.status) + return + } + http.NotFound(w, r) + })) + defer server.Close() + + client, err := pbs.NewClient(pbs.ClientConfig{ + Host: server.URL, + TokenName: "root@pam!token", + TokenValue: "secret", + }) + if err != nil { + t.Fatalf("failed to create PBS client: %v", err) + } + + m := &Monitor{state: models.NewState()} + original := models.PBSBackup{ + ID: "pbs-pbs1-archive--vm-100-1700000000", + Instance: "pbs1", + Datastore: "archive", + Namespace: "", + BackupType: "vm", + VMID: "100", + BackupTime: time.Unix(1700000000, 0), + } + m.state.UpdatePBSBackups("pbs1", []models.PBSBackup{original}) + + m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ + {Name: "archive"}, + }) + + snapshot := m.state.GetSnapshot() + var found bool + for _, backup := range snapshot.PBSBackups { + if backup.Instance == "pbs1" && backup.ID == original.ID { + found = true + break + } + } + if !found { + t.Fatal("expected cached backup to be preserved on transient error") + } + }) + } +} diff --git a/internal/monitoring/monitor_pbs_coverage_test.go b/internal/monitoring/monitor_pbs_coverage_test.go index d008a8209..038076ce4 100644 --- a/internal/monitoring/monitor_pbs_coverage_test.go +++ b/internal/monitoring/monitor_pbs_coverage_test.go @@ -11,6 +11,7 @@ import ( "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/pkg/pbs" @@ -500,3 +501,70 @@ func TestPBSAndPMGPollSkipDisabledInstances(t *testing.T) { } } } + +// Exercise real HTTP polling and one manager through missing metrics and recovery. +func TestPBSMetricAvailabilityAlertLifecycle(t *testing.T) { + fixture := newPBSHealthTestServer(t) + instance := config.PBSInstance{Name: "pbs-lifecycle", Host: fixture.server.URL, MonitorDatastores: true} + monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance}) + client := newPBSHealthTestClient(t, instance.Host) + manager := alerts.NewManagerWithDataDir(t.TempDir()) + defer manager.Stop() + monitor.alertManager = manager + manager.UpdateConfig(alerts.AlertConfig{Enabled: true, ActivationState: alerts.ActivationActive, + TimeThresholds: map[string]int{"pbs": 0}, PBSDefaults: alerts.ThresholdConfig{ + Memory: &alerts.HysteresisThreshold{Trigger: 40, Clear: 30}, + }}) + fired, resolved := make(chan string, 8), make(chan string, 8) + manager.SetAlertCallback(func(a *alerts.Alert) { fired <- a.ID }) + manager.SetResolvedCallback(func(id string) { resolved <- id }) + poll := func() models.PBSInstance { + monitor.pollPBSInstance(context.Background(), instance.Name, client) + return pbsInstanceByName(t, monitor.state.GetSnapshot(), instance.Name) + } + poll() + if len(manager.GetActiveAlerts()) != 1 { + t.Fatal("high memory failed to activate") + } + select { + case <-fired: + case <-time.After(time.Second): + t.Fatal("missing alert dispatch") + } + for _, mode := range []pbsHealthTestMode{pbsHealthTestNodeDenied, pbsHealthTestNodeGatewayFailure} { + fixture.setMode(mode) + missing := poll() + if missing.Status != "online" || !missing.NodeMetricsUnavailable { + t.Fatalf("bad missing projection: %+v", missing) + } + for range 5 { + poll() + } + if len(manager.GetActiveAlerts()) != 1 { + t.Fatal("endpoint failure resolved memory alert") + } + if len(manager.GetRecentlyResolved()) != 0 { + t.Fatal("false resolved history") + } + select { + case id := <-resolved: + t.Fatalf("false recovery dispatch %s", id) + case <-time.After(50 * time.Millisecond): + } + } + fixture.setMode(pbsHealthTestLowMemory) + for range 5 { + poll() + } + if len(manager.GetActiveAlerts()) != 0 { + t.Fatal("valid low memory failed to recover") + } + select { + case <-resolved: + case <-time.After(time.Second): + t.Fatal("missing recovery dispatch") + } + if len(manager.GetRecentlyResolved()) != 1 { + t.Fatal("missing resolved history") + } +} diff --git a/internal/monitoring/monitor_pbs_health_authority_test.go b/internal/monitoring/monitor_pbs_health_authority_test.go index a7f20894a..117be78f4 100644 --- a/internal/monitoring/monitor_pbs_health_authority_test.go +++ b/internal/monitoring/monitor_pbs_health_authority_test.go @@ -22,6 +22,11 @@ const ( pbsHealthTestAuthFailure pbsHealthTestTimeout pbsHealthTestPartialData + pbsHealthTestNodeDenied + pbsHealthTestNodeGatewayFailure + pbsHealthTestUnavailable + pbsHealthTestLowMemory + pbsHealthTestNullNodeStatus ) type pbsHealthTestServer struct { @@ -47,6 +52,9 @@ func newPBSHealthTestServer(t *testing.T) *pbsHealthTestServer { case pbsHealthTestAuthFailure: http.Error(w, "authentication failed: 401 Unauthorized", http.StatusUnauthorized) return + case pbsHealthTestUnavailable: + http.Error(w, "service unavailable", http.StatusServiceUnavailable) + return case pbsHealthTestTimeout: <-r.Context().Done() return @@ -59,10 +67,27 @@ func newPBSHealthTestServer(t *testing.T) *pbsHealthTestServer { "data": map[string]any{"version": "3.4.2"}, }) case "/api2/json/nodes/localhost/status": + if mode == pbsHealthTestNullNodeStatus { + _, _ = w.Write([]byte(`{"data":null}`)) + return + } + if mode == pbsHealthTestNodeDenied { + http.Error(w, "permission denied", http.StatusForbidden) + return + } + if mode == pbsHealthTestNodeGatewayFailure { + http.Error(w, "gateway unavailable", http.StatusBadGateway) + return + } _ = json.NewEncoder(w).Encode(map[string]any{ "data": map[string]any{ - "cpu": 0.15, - "memory": map[string]any{"used": 512, "total": 1024}, + "cpu": 0.15, + "memory": map[string]any{"used": func() int { + if mode == pbsHealthTestLowMemory { + return 100 + } + return 512 + }(), "total": 1024}, "uptime": 120, }, }) @@ -297,3 +322,69 @@ func TestInitPBSClientsDoesNotTreatClientConstructionAsConnectivity(t *testing.T } assertPBSConnectionProjection(t, monitor, "pbs-invalid-url", false, "offline") } + +// TestPollPBSNodeMetricsFailureAndRecovery characterises connectivity separately +// from metric availability. Zero-valued metrics are the current projection of +// unavailable data, not evidence of measured zero utilisation or alert recovery. +func TestPollPBSNodeMetricsFailureAndRecovery(t *testing.T) { + for _, tc := range []struct { + name string + mode pbsHealthTestMode + }{ + {"permission-denied", pbsHealthTestNodeDenied}, + {"endpoint-gateway-failure", pbsHealthTestNodeGatewayFailure}, + } { + t.Run(tc.name, func(t *testing.T) { + fixture := newPBSHealthTestServer(t) + instance := config.PBSInstance{Name: "pbs-metrics", Host: fixture.server.URL, MonitorDatastores: true} + monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance}) + client := newPBSHealthTestClient(t, instance.Host) + poll := func() models.PBSInstance { + monitor.pollPBSInstance(context.Background(), instance.Name, client) + return pbsInstanceByName(t, monitor.state.GetSnapshot(), instance.Name) + } + assertMetrics := func(got models.PBSInstance) { + t.Helper() + if got.NodeMetricsUnavailable { + t.Fatal("successful metrics marked unavailable") + } + if got.CPU != 0.15 || got.Memory != 50 || got.MemoryUsed != 512 || got.MemoryTotal != 1024 || got.Uptime != 120 { + t.Fatalf("successful node metrics not published: %+v", got) + } + } + assertMetrics(poll()) + previous := monitor.pollStatusMap["pbs::"+instance.Name].LastSuccess + fixture.setMode(tc.mode) + partial := poll() + assertPBSConnectionProjection(t, monitor, instance.Name, true, "online") + status := monitor.pollStatusMap["pbs::"+instance.Name] + if !status.LastSuccess.After(previous) || status.ConsecutiveFailures != 0 || status.LastErrorMessage != "" { + t.Fatalf("endpoint failure incorrectly affected connectivity ledger: %+v", status) + } + if !partial.NodeMetricsUnavailable { + t.Fatal("failed node endpoint not marked unavailable") + } + if partial.CPU != 0 || partial.Memory != 0 || partial.MemoryUsed != 0 || partial.MemoryTotal != 0 || partial.Uptime != 0 { + t.Fatalf("unavailable node metrics retained previous measurements: %+v", partial) + } + if len(partial.Datastores) != 1 || partial.Datastores[0].Name != "backups" { + t.Fatalf("node endpoint failure discarded accessible datastore: %+v", partial.Datastores) + } + fixture.setMode(pbsHealthTestUnavailable) + lastConnected := status.LastSuccess + poll() + assertPBSConnectionProjection(t, monitor, instance.Name, false, "offline") + status = monitor.pollStatusMap["pbs::"+instance.Name] + if !status.LastSuccess.Equal(lastConnected) || status.ConsecutiveFailures != 1 || status.LastErrorMessage == "" { + t.Fatalf("full outage not recorded independently: %+v", status) + } + fixture.setMode(pbsHealthTestSuccess) + assertMetrics(poll()) + assertPBSConnectionProjection(t, monitor, instance.Name, true, "online") + status = monitor.pollStatusMap["pbs::"+instance.Name] + if !status.LastSuccess.After(lastConnected) || status.ConsecutiveFailures != 0 || status.LastErrorMessage != "" || !status.LastErrorAt.IsZero() { + t.Fatalf("recovery did not clear current connectivity failure: %+v", status) + } + }) + } +} diff --git a/internal/monitoring/monitor_pbs_pmg.go b/internal/monitoring/monitor_pbs_pmg.go index 6021ed993..01f6e3ae7 100644 --- a/internal/monitoring/monitor_pbs_pmg.go +++ b/internal/monitoring/monitor_pbs_pmg.go @@ -241,14 +241,15 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie // Initialize PBS instance with default values pbsInst = models.PBSInstance{ - ID: PBSMonitorResourceID(instanceName), - Name: instanceName, - Host: instanceCfg.Host, - GuestURL: instanceCfg.GuestURL, - Status: "offline", - Version: "unknown", - ConnectionHealth: "unhealthy", - LastSeen: time.Now(), + NodeMetricsUnavailable: true, + ID: PBSMonitorResourceID(instanceName), + Name: instanceName, + Host: instanceCfg.Host, + GuestURL: instanceCfg.GuestURL, + Status: "offline", + Version: "unknown", + ConnectionHealth: "unhealthy", + LastSeen: time.Now(), } publishResult = true @@ -336,6 +337,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie log.Debug().Err(err).Str("instance", instanceName).Msg("could not get PBS node status (may need Sys.Audit permission)") } } else if nodeStatus != nil { + pbsInst.NodeMetricsUnavailable = false pbsInst.CPU = nodeStatus.CPU if nodeStatus.Memory.Total > 0 { pbsInst.Memory = float64(nodeStatus.Memory.Used) / float64(nodeStatus.Memory.Total) * 100 diff --git a/internal/monitoring/monitor_pbs_webhook_test.go b/internal/monitoring/monitor_pbs_webhook_test.go new file mode 100644 index 000000000..098ae51a8 --- /dev/null +++ b/internal/monitoring/monitor_pbs_webhook_test.go @@ -0,0 +1,144 @@ +package monitoring + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "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/notifications" +) + +// This is local integration proof, not installed-artifact or off-host delivery +// qualification. Both the synthetic PBS and receiver live in this test process. +func TestPBSPartialMetricsWebhookLifecycle(t *testing.T) { + type payload struct { + Event string `json:"event"` + Alerts []alerts.Alert `json:"alerts"` + } + received := make(chan payload, 16) + receiver := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var p payload + if r.Method != http.MethodPost || json.NewDecoder(r.Body).Decode(&p) != nil { + t.Error("invalid webhook request") + http.Error(w, "invalid request", http.StatusBadRequest) + return + } + select { + case received <- p: + default: + t.Error("unexpected webhook flood") + } + w.WriteHeader(http.StatusOK) + })) + defer receiver.Close() + nm := notifications.NewNotificationManagerWithDataDir("", t.TempDir()) + defer nm.Stop() + nm.SetGroupingWindow(0) + nm.SetNotifyOnResolve(true) + if err := nm.UpdateAllowedPrivateCIDRs("127.0.0.1/32"); err != nil { + t.Fatal(err) + } + nm.AddWebhook(notifications.WebhookConfig{ID: "local-receiver", Name: "local-receiver", URL: receiver.URL, Enabled: true}) + + fixture := newPBSHealthTestServer(t) + instance := config.PBSInstance{Name: "pbs-webhook", Host: fixture.server.URL, MonitorDatastores: true} + monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance}) + manager := alerts.NewManagerWithDataDir(t.TempDir()) + defer manager.Stop() + monitor.alertManager, monitor.notificationMgr = manager, nm + manager.UpdateConfig(alerts.AlertConfig{Enabled: true, ActivationState: alerts.ActivationActive, + TimeThresholds: map[string]int{"pbs": 0}, PBSDefaults: alerts.ThresholdConfig{ + Memory: &alerts.HysteresisThreshold{Trigger: 40, Clear: 30}, + }}) + monitor.wireExternalAlertCallbacks(nil) + client := newPBSHealthTestClient(t, instance.Host) + poll := func() { monitor.pollPBSInstance(context.Background(), instance.Name, client) } + receive := func() payload { + t.Helper() + select { + case p := <-received: + return p + case <-time.After(5 * time.Second): + t.Fatal("webhook not received") + return payload{} + } + } + // Wait for the queue's completed delivery audit, not merely an HTTP request: + // recovery eligibility is recorded after the receiver responds successfully. + waitSent := func(want int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + entries, err := nm.GetDeliveryLog(time.Time{}, 20) + if err != nil { + t.Fatal(err) + } + sent := 0 + for _, e := range entries { + if e.Success { + sent++ + } + } + if sent == want { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("did not record %d successful deliveries", want) + } + poll() + firing := receive() + if len(firing.Alerts) != 1 || firing.Alerts[0].Type != "memory" || firing.Event == "resolved" { + t.Fatalf("unexpected firing payload: %+v", firing) + } + incident := firing.Alerts[0] + waitSent(1) + for _, mode := range []pbsHealthTestMode{pbsHealthTestNodeDenied, pbsHealthTestNodeGatewayFailure, pbsHealthTestNullNodeStatus} { + fixture.setMode(mode) + for range 5 { + poll() + } + projection := pbsInstanceByName(t, monitor.state.GetSnapshot(), instance.Name) + if projection.Status != "online" || !projection.NodeMetricsUnavailable { + t.Fatalf("partial failure must retain connectivity without claiming metrics: %+v", projection) + } + active := manager.GetActiveAlerts() + if len(active) != 1 || active[0].ID != incident.ID || !active[0].StartTime.Equal(incident.StartTime) { + t.Fatal("partial failure changed the active incident") + } + if len(manager.GetRecentlyResolved()) != 0 { + t.Fatal("partial failure fabricated recovery history") + } + select { + case p := <-received: + t.Fatalf("partial failure sent webhook: %+v", p) + case <-time.After(150 * time.Millisecond): + } + } + fixture.setMode(pbsHealthTestLowMemory) + for range 5 { + poll() + } + recovery := receive() + if recovery.Event != "resolved" || len(recovery.Alerts) != 1 || + recovery.Alerts[0].ID != incident.ID || !recovery.Alerts[0].StartTime.Equal(incident.StartTime) { + t.Fatalf("recovery does not identify delivered incident: %+v", recovery) + } + waitSent(2) + if len(manager.GetActiveAlerts()) != 0 || len(manager.GetRecentlyResolved()) != 1 { + t.Fatal("genuine recovery did not update active/history state") + } + for range 5 { + poll() + } + select { + case p := <-received: + t.Fatalf("duplicate webhook after recovery: %+v", p) + case <-time.After(150 * time.Millisecond): + } +} diff --git a/internal/monitoring/pbs_backup_cache_terminal_test.go b/internal/monitoring/pbs_backup_cache_terminal_test.go deleted file mode 100644 index 8b01baca6..000000000 --- a/internal/monitoring/pbs_backup_cache_terminal_test.go +++ /dev/null @@ -1,109 +0,0 @@ -package monitoring - -import ( - "context" - "net/http" - "net/http/httptest" - "strings" - "testing" - "time" - - "github.com/rcourtman/pulse-go-rewrite/internal/models" - "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" -) - -func TestPollPBSBackups_DropsStaleCacheOnTerminalDatastoreError(t *testing.T) { - t.Parallel() - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if strings.Contains(r.URL.Path, "/admin/datastore/archive/groups") { - http.Error(w, `{"errors":"datastore does not exist"}`, http.StatusNotFound) - return - } - http.NotFound(w, r) - })) - defer server.Close() - - client, err := pbs.NewClient(pbs.ClientConfig{ - Host: server.URL, - TokenName: "root@pam!token", - TokenValue: "secret", - }) - if err != nil { - t.Fatalf("failed to create PBS client: %v", err) - } - - m := &Monitor{state: models.NewState()} - m.state.UpdatePBSBackups("pbs1", []models.PBSBackup{ - { - ID: "pbs-pbs1-archive--vm-100-1700000000", - Instance: "pbs1", - Datastore: "archive", - Namespace: "", - BackupType: "vm", - VMID: "100", - BackupTime: time.Unix(1700000000, 0), - }, - }) - - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ - {Name: "archive"}, - }) - - snapshot := m.state.GetSnapshot() - for _, backup := range snapshot.PBSBackups { - if backup.Instance == "pbs1" { - t.Fatalf("expected stale backups to be removed after terminal error, found: %+v", backup) - } - } -} - -func TestPollPBSBackups_PreservesCacheOnTransientDatastoreError(t *testing.T) { - t.Parallel() - - server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if strings.Contains(r.URL.Path, "/admin/datastore/archive/groups") { - http.Error(w, `{"errors":"temporary server issue"}`, http.StatusInternalServerError) - return - } - http.NotFound(w, r) - })) - defer server.Close() - - client, err := pbs.NewClient(pbs.ClientConfig{ - Host: server.URL, - TokenName: "root@pam!token", - TokenValue: "secret", - }) - if err != nil { - t.Fatalf("failed to create PBS client: %v", err) - } - - m := &Monitor{state: models.NewState()} - original := models.PBSBackup{ - ID: "pbs-pbs1-archive--vm-100-1700000000", - Instance: "pbs1", - Datastore: "archive", - Namespace: "", - BackupType: "vm", - VMID: "100", - BackupTime: time.Unix(1700000000, 0), - } - m.state.UpdatePBSBackups("pbs1", []models.PBSBackup{original}) - - m.pollPBSBackups(context.Background(), "pbs1", client, []models.PBSDatastore{ - {Name: "archive"}, - }) - - snapshot := m.state.GetSnapshot() - var found bool - for _, backup := range snapshot.PBSBackups { - if backup.Instance == "pbs1" && backup.ID == original.ID { - found = true - break - } - } - if !found { - t.Fatal("expected cached backup to be preserved on transient error") - } -} diff --git a/pkg/pbs/client.go b/pkg/pbs/client.go index 6e1844465..ef9aba61e 100644 --- a/pkg/pbs/client.go +++ b/pkg/pbs/client.go @@ -331,7 +331,9 @@ func (e *apiHTTPError) Error() string { return message } -func pbsHTTPStatus(err error) (int, bool) { +// HTTPStatus returns the response status from a PBS API or authentication error, +// including wrapped errors. The response body is never used for classification. +func HTTPStatus(err error) (int, bool) { var apiErr *apiHTTPError if errors.As(err, &apiErr) { return apiErr.status, true @@ -346,12 +348,12 @@ func pbsHTTPStatus(err error) (int, bool) { } func isPBSPermissionError(err error) bool { - status, ok := pbsHTTPStatus(err) + status, ok := HTTPStatus(err) return ok && (status == http.StatusUnauthorized || status == http.StatusForbidden) } func isPBSNotFoundError(err error) bool { - status, ok := pbsHTTPStatus(err) + status, ok := HTTPStatus(err) return ok && status == http.StatusNotFound } @@ -769,14 +771,20 @@ func (c *Client) GetNodeStatus(ctx context.Context) (*NodeStatus, error) { log.Debug().Str("response", string(body)).Msg("PBS node status response") var statusResult struct { - Data NodeStatus `json:"data"` + Data *NodeStatus `json:"data"` } if err := json.Unmarshal(body, &statusResult); err != nil { return nil, fmt.Errorf("failed to decode status response: %w", err) } - return &statusResult.Data, nil + // A successful HTTP response is not evidence of available metrics. + // Missing/null data must not become zero usage and resolve active alerts. + if statusResult.Data == nil { + return nil, fmt.Errorf("node status response contains no data") + } + + return statusResult.Data, nil } // GetDatastores returns all datastores with their status. diff --git a/pkg/pbs/client_http_test.go b/pkg/pbs/client_http_test.go index cb2dfa0a0..68eba0500 100644 --- a/pkg/pbs/client_http_test.go +++ b/pkg/pbs/client_http_test.go @@ -580,7 +580,7 @@ func TestClient_GetNodeName_SuperuserPermissionFailureRetries(t *testing.T) { if firstErr == nil { t.Fatal("first GetNodeName: expected error") } - if got, ok := pbsHTTPStatus(firstErr); !ok || got != status { + if got, ok := HTTPStatus(firstErr); !ok || got != status { t.Fatalf("first GetNodeName status = (%d, %v), want (%d, true): %v", got, ok, status, firstErr) } name, err := client.GetNodeName(context.Background()) @@ -632,7 +632,7 @@ func TestClient_GetNodeName_TransientHTTPFailuresRetryAndRecover(t *testing.T) { if firstErr == nil { t.Fatal("first GetNodeName: expected error") } - if got, ok := pbsHTTPStatus(firstErr); !ok || got != tc.status { + if got, ok := HTTPStatus(firstErr); !ok || got != tc.status { t.Fatalf("first GetNodeName status = (%d, %v), want (%d, true): %v", got, ok, tc.status, firstErr) } name, err := client.GetNodeName(context.Background()) @@ -795,7 +795,7 @@ func TestClient_GetNodeName_ConcurrentTransientFailureIsSingleFlight(t *testing. if err == nil { t.Fatal("concurrent GetNodeName: expected transient error") } - if got, ok := pbsHTTPStatus(err); !ok || got != http.StatusServiceUnavailable { + if got, ok := HTTPStatus(err); !ok || got != http.StatusServiceUnavailable { t.Fatalf("concurrent GetNodeName status = (%d, %v), want (503, true): %v", got, ok, err) } } @@ -817,3 +817,18 @@ func TestClient_GetNodeName_ConcurrentTransientFailureIsSingleFlight(t *testing. t.Fatalf("/nodes hit %d times after recovery and cache read, want 2", got) } } +func TestClient_GetNodeStatus_MissingData(t *testing.T) { + for _, body := range []string{`{"data":null}`, `{}`, `null`} { + t.Run(body, func(t *testing.T) { + client, server := newTestClient(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(body)) + }) + defer server.Close() + status, err := client.GetNodeStatus(context.Background()) + if status != nil || err == nil { + t.Fatalf("absent metrics = (%+v, %v), want nil status and error", status, err) + } + }) + } +} diff --git a/tests/qualification/navigation-reconnect/README.md b/tests/qualification/navigation-reconnect/README.md index a6fbaf428..6a248a0ec 100644 --- a/tests/qualification/navigation-reconnect/README.md +++ b/tests/qualification/navigation-reconnect/README.md @@ -40,3 +40,12 @@ binds this run to the release-line base and runtime source hashes. Existing critical-transition and stable-identity recovery tests also passed three race repetitions on this base with the backport applied. + +Follow-up hook coverage on 5 September separates the mocked resource-snapshot +flag from general hydration. Alert-only hydration now explicitly reports +initialDataReceived=true with resourceSnapshotReceived=false; the inverse case +retains an authoritative empty resource snapshot during reconnect. All 88 tests +in the three focused files above pass. Temporarily substituting +initialDataReceived for resourceSnapshotReceived in runtimeStateResolved makes +both boundary tests fail (two failures, 21 skipped); the mutation was removed. +This verifies regression sensitivity, not installed browser or soak readiness.