From 4815a5c80d7e94d21ff14002986b47ce23dd0be5 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 24 Jul 2026 09:45:00 +0100 Subject: [PATCH] Fix PBS connection health authority --- ...-for-ga-pbs-health-authority-2026-07-24.md | 92 ++++++ docs/release-control/v6/internal/status.json | 28 +- .../v6/internal/subsystems/agent-lifecycle.md | 9 + .../v6/internal/subsystems/api-contracts.md | 11 + .../subsystems/frontend-primitives.md | 5 + .../v6/internal/subsystems/monitoring.md | 15 + .../internal/subsystems/storage-recovery.md | 12 + .../Settings/DiagnosticsResultsPanel.tsx | 34 +- .../DiagnosticsResultsPanel.test.tsx | 50 +++ .../__tests__/settingsArchitecture.test.ts | 2 + .../components/Settings/diagnosticsModel.ts | 10 + .../api/config_handlers_connection_test.go | 45 +++ internal/api/connections_aggregator.go | 6 +- internal/api/connections_aggregator_test.go | 81 +++++ internal/api/diagnostics.go | 81 ++++- internal/api/diagnostics_additional_test.go | 8 +- internal/api/diagnostics_test.go | 52 +++ .../monitoring/canonical_guardrails_test.go | 16 + internal/monitoring/monitor_client_init.go | 21 +- .../monitoring/monitor_client_reconnect.go | 20 +- .../monitoring/monitor_full_coverage_test.go | 1 + .../monitoring/monitor_pbs_coverage_test.go | 6 + .../monitor_pbs_health_authority_test.go | 299 ++++++++++++++++++ internal/monitoring/monitor_pbs_pmg.go | 130 +++++--- 24 files changed, 944 insertions(+), 90 deletions(-) create mode 100644 docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md create mode 100644 internal/monitoring/monitor_pbs_health_authority_test.go diff --git a/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md b/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md new file mode 100644 index 0000000000..1e31b017f4 --- /dev/null +++ b/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md @@ -0,0 +1,92 @@ +# Known RC Issue Closure For GA PBS Health Authority Record + +- Date: `2026-07-24` +- Gate: `known-rc-issue-closure-for-ga` +- Issue: `#1465` +- Result: `fixed-main-proof` + +## Context + +Issue `#1465` showed two configured PBS systems as Active/Fleet OK in Settings +while the Dashboard rendered both as errors and the saved connection test +failed. The original `v6.0.0-rc.4` cause included deferred poll accounting that +captured a nil error before the poll ran. Commits `bf6261adc` and `bd7d196c1` +corrected that capture and are ancestors of `v6.1.1`. + +Release-tag replay proved that correction was necessary but incomplete. +`v6.1.1` records an authentication failure in the scheduler ledger, yet the +authentication branch returns before updating the Dashboard PBS model. +Separately, the connections aggregator treats a new timeout as Active whenever +an earlier success exists and the circuit breaker has not opened. The same +current error is already defined as outstanding state and is cleared on the +next success, so the Active projection contradicts both the Dashboard and the +ledger. + +## Disposition + +PBS polling now finalizes one dynamic poll error and projects that result into +the scheduler ledger, staleness tracking, poll metrics, connection-health map, +Dashboard PBS model, and legacy PBS alert evaluation. Authentication, timeout, +cancellation, initialization, and panic failures publish offline/error +consistently. Optional node, datastore, namespace, and job collection failures +remain partial data evidence and do not convert proven connectivity into a +connection failure. + +Client creation and retry recreation no longer claim connectivity before a +poll. Invalid initialization records and publishes a failed outcome; successful +construction remains pending until the first completed poll. + +The connections API now treats any current `LastError` as unreachable, or +unauthorized for authentication errors, even when `LastSuccess` is preserved +for freshness context. Settings Fleet governance and connection alerts consume +that same connection row. Diagnostics exposes the canonical monitored state +and the immediate support probe separately, and the Settings diagnostics row +names both when they disagree. Saved Test Connection remains an isolated +credential/network probe and does not mutate runtime health. + +## Scenario Proof + +The managed dual-PBS HTTP fixture and API projection tests cover: + +- two independently keyed PBS systems sharing the same hostname; +- initial success followed by authentication rejection and network timeout; +- preservation of last-success freshness alongside the current failure; +- a successful version probe with failed datastore inventory as connected + partial data; +- credential rotation, URL replacement, monitor restart, and recovery; +- invalid URL initialization and first-poll pending behavior; +- consistent Settings/API/Fleet, Dashboard, alert-snapshot, and diagnostics + projections; +- a direct saved-connection test using the current stored token without + changing scheduler or connection health. + +## Proof + +- `v6.1.1`: `TestMonitor_PollPBSInstance_AuthFailure` passes, proving the RC5 + deferred-accounting correction is present. +- `v6.1.1` with the new regressions applied: the authentication case leaves + the Dashboard model online, and a post-success timeout plus the same-hostname + second PBS row both remain Active. These expected failures reproduce the + unresolved contradiction on the release tag. +- `go test ./internal/monitoring -count=1` +- `go test ./internal/api -count=1` +- `go test -race ./internal/monitoring -run + 'TestPollPBSInstancesKeepsAllHealthProjectionsOnOneOutcome|TestPollPBSInstanceKeepsPartialDataSeparateFromConnectivity|TestInitPBSClientsDoesNotTreatClientConstructionAsConnectivity|TestMonitor_PollPBSInstance_AuthFailure' + -count=1` +- `go test -race ./internal/api -run + 'TestDeriveConnectionState_CurrentFailureOverridesRecentSuccess|TestBuildConnectionsKeepsDistinctPBSSystemsWithSameHostname|TestPBSDiagnosticsKeepsCanonicalPollStateSeparateFromLiveProbe|TestHandleTestNodePBSIsADirectProbeWithoutChangingRuntimeHealth' + -count=1` +- focused Settings diagnostics, connections-ledger, infrastructure-source, and + architecture Vitest suites: 169 tests passed +- frontend TypeScript, ESLint, Prettier, and Settings diagnostics-boundary + checks +- subsystem registry, contract, status, control-plane, staged-shape, and + canonical-completion audits + +## Outcome + +The RC5 deferred-accounting correction is present in `v6.1.1`, but the complete +cross-surface behavior in this record is not. The remaining source-authority +fix and regressions are on `main` for a future v6 release; no backport or +publication date is claimed. Issue `#1465` remains open for reporter +confirmation after retesting a release that contains this change. diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index a1c6625bac..128f81f1c6 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -5742,6 +5742,12 @@ "kind": "file", "evidence_tier": "managed-runtime-exercise" }, + { + "repo": "pulse", + "path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md", + "kind": "file", + "evidence_tier": "test-proof" + }, { "repo": "pulse", "path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-proxmox-cluster-address-authority-2026-07-23.md", @@ -7819,6 +7825,12 @@ "kind": "file", "evidence_tier": "managed-runtime-exercise" }, + { + "repo": "pulse", + "path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-pbs-health-authority-2026-07-24.md", + "kind": "file", + "evidence_tier": "test-proof" + }, { "repo": "pulse", "path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-proxmox-cluster-address-authority-2026-07-23.md", @@ -9194,7 +9206,21 @@ ] } ], - "work_claims": [], + "work_claims": [ + { + "id": "issue-1124-metrics-write-amplification-lane-l10", + "agent_id": "issue-1124-metrics-write-amplification", + "summary": "Measure and reduce metrics.db physical write amplification for issue #1124 without weakening durability or bounded runtime behavior.", + "target_id": "v6-product-lane-expansion", + "claimed_at": "2026-07-24T08:05:22Z", + "heartbeat_at": "2026-07-24T08:05:22Z", + "expires_at": "2026-07-24T16:05:22Z", + "work_item": { + "kind": "lane", + "id": "L10" + } + } + ], "open_decisions": [], "source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md", "resolved_decisions": [ diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index e169860e3b..96e744c5f5 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -2101,6 +2101,15 @@ Agent` secondary handoff against the live setup wizard instead of relying ## Current State +### PBS connection health does not create agent lifecycle evidence + +The shared API connections ledger and diagnostics now project PBS health from +the monitoring scheduler's completed poll result and keep immediate diagnostics +probes separate. This does not enroll a PBS host as a Pulse Agent, extend an +agent heartbeat, prove command reachability, or create update/repair authority. +Agent lifecycle consumers continue to use agent-source identity and report +freshness only; same-hostname PBS connections remain distinct provider rows. + ### Docker and Podman report sizes share one exact-byte contract The Unified Agent now measures the encoded gzip body and decoded report JSON diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 4ce2166ca7..fd8add889d 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -5916,6 +5916,17 @@ machine-readable `errorKind` values with operator guidance in certificate mismatch, refused connections, unreachable networks, DNS failures, timeouts, and unknown connection failures must remain distinguishable in support bundles instead of collapsing into a generic connection failure string. +PBS connection state follows the same scheduler-ledger authority as the +Settings connections response and connection-alert snapshots. +`InstancePollStatus.LastError` is a current outstanding failure that is cleared +by the next successful poll; a preserved `LastSuccess` supplies freshness +context but must not downgrade that current failure back to `active` before the +circuit breaker opens. `/api/diagnostics` exposes that canonical state through +`PBSDiagnostic.connected`, `state`, `stateReason`, `lastSeen`, and `lastError`. +Its immediate credential/network check is separate `probe` evidence, so a +successful support probe can demonstrate recovery in progress without +overwriting the last completed runtime poll, and a transient probe failure +cannot make a healthy monitored source appear disconnected. That same diagnostics boundary now explicitly excludes maintainer analytics. `internal/api/diagnostics.go` must not serialize commercial funnel, sales funnel, pricing/checkout conversion, or infrastructure onboarding telemetry in diff --git a/docs/release-control/v6/internal/subsystems/frontend-primitives.md b/docs/release-control/v6/internal/subsystems/frontend-primitives.md index 5383bfb3c6..f08a41b841 100644 --- a/docs/release-control/v6/internal/subsystems/frontend-primitives.md +++ b/docs/release-control/v6/internal/subsystems/frontend-primitives.md @@ -3866,6 +3866,11 @@ run/export lifecycle, results rendering, sanitization/model helpers, and customer-facing diagnostics copy. The shell must not re-accumulate inline API calls, export-download plumbing, diagnostics-card composition, or diagnostics surface copy. +PBS diagnostics status badges render the backend's canonical monitored +`connected`/`state` result. The one-off diagnostics request remains a nested +`probe` result; when the probe and monitor disagree, the row must name both +states instead of turning a successful live check into a green monitored badge +or hiding recovery evidence behind an undifferentiated failure. That same diagnostics owner split also keeps maintainer analytics out of the customer diagnostics surface. `DiagnosticsResultsPanel.tsx`, `diagnosticsModel.ts`, and the diagnostics export path must not render or diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 3ce0441215..a595a8cf27 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -644,6 +644,21 @@ node-local Agent evidence. ## Current State +### PBS health is one completed-poll outcome + +PBS client construction is transport setup, not connectivity evidence. Initial +client creation and retry recreation therefore remain pending until a poll +completes; client-construction failures publish a failed scheduler result +instead of leaving Settings pending while legacy state is disconnected. +`pollPBSInstance` finalizes its dynamic `pollErr` once and uses that outcome for +the scheduler ledger, staleness tracker, poll metrics, connection-health map, +dashboard `PBSInstance`, and legacy PBS alert evaluation. Authentication, +timeout, cancellation, and panic outcomes all publish `offline`/`error`, while +a later success clears the current error and publishes `online`/`healthy`. +Optional node, datastore, namespace, or job collection failures remain partial +data evidence and do not turn a successful version/datastore connectivity +probe into a connection failure. + ### Host snapshots carry integration provenance; doctor copy is user-facing `models.Host.IntegrationSource` mirrors the unified fabric's diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 5007a56e7e..125f6c028c 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -1950,6 +1950,18 @@ registered-agent counts does not alter any Recovery Assurance domain: recovery evidence still comes exclusively from its own deterministic verification paths. +### PBS reachability remains separate from recovery freshness + +The shared PBS scheduler outcome may mark a source unreachable or unauthorized, +but storage and recovery consumers must not reinterpret that connection state +as deletion, corruption, or current freshness of previously collected +datastores, backups, or recovery points. Conversely, a successful PBS version +or datastore connectivity probe proves reachability only; optional inventory +and job collection can remain partial, and Recovery Assurance continues to +derive protection and recoverability from its own timestamped evidence. +`internal/api/connections_aggregator.go` and diagnostics may project the shared +connection outcome, but they do not become storage or restore authority. + ### Recovery actions retain execution-time readiness checks Recovery-oriented capabilities admitted to canonical Actions are subject to diff --git a/frontend-modern/src/components/Settings/DiagnosticsResultsPanel.tsx b/frontend-modern/src/components/Settings/DiagnosticsResultsPanel.tsx index d76417f480..c94b80dd32 100644 --- a/frontend-modern/src/components/Settings/DiagnosticsResultsPanel.tsx +++ b/frontend-modern/src/components/Settings/DiagnosticsResultsPanel.tsx @@ -172,16 +172,30 @@ export const DiagnosticsResultsPanel: Component =
{(pbs) => ( -
- - {pbs.name} - - +
+
+ + {pbs.name} + + +
+ + {(probe) => ( +
+ Monitor: {pbs.state || (pbs.connected ? 'active' : 'unreachable')}; live + check: {probe().connected ? 'connected' : 'failed'} +
+ )} +
)} diff --git a/frontend-modern/src/components/Settings/__tests__/DiagnosticsResultsPanel.test.tsx b/frontend-modern/src/components/Settings/__tests__/DiagnosticsResultsPanel.test.tsx index 6d17a3635b..405d90a216 100644 --- a/frontend-modern/src/components/Settings/__tests__/DiagnosticsResultsPanel.test.tsx +++ b/frontend-modern/src/components/Settings/__tests__/DiagnosticsResultsPanel.test.tsx @@ -244,4 +244,54 @@ describe('DiagnosticsResultsPanel', () => { ); expect(screen.queryByText('Container Runtime Agents')).not.toBeInTheDocument(); }); + + it('shows direct PBS probe evidence without replacing monitored state', () => { + const diagnosticsData = { + version: '6.1.1', + runtime: 'go', + uptime: 3600, + nodes: [], + pbs: [ + { + id: 'pbs-primary', + name: 'pbs-primary', + host: 'https://backup.local:8007', + connected: false, + state: 'unreachable', + stateReason: 'connection timed out', + probe: { + connected: true, + details: { version: '3.4.2' }, + }, + }, + ], + system: { + os: 'linux', + arch: 'amd64', + goVersion: 'go1.25', + numCPU: 8, + numGoroutine: 32, + memoryMB: 128, + }, + errors: [], + } as DiagnosticsData; + + render(() => ( + + ( + {}} + /> + )} + /> + + )); + + expect(screen.getByText('pbs-primary')).toBeInTheDocument(); + expect(screen.getByText('Monitor: unreachable; live check: connected')).toBeInTheDocument(); + }); }); diff --git a/frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts b/frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts index 5009b4c289..bee0dc0855 100644 --- a/frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts +++ b/frontend-modern/src/components/Settings/__tests__/settingsArchitecture.test.ts @@ -1700,6 +1700,8 @@ describe('settings architecture guardrails', () => { expect(diagnosticsModelSource).toContain('stripInternalAnalyticsDiagnosticsFields'); expect(diagnosticsModelSource).toContain('assistantRuntimeConnected'); + expect(diagnosticsModelSource).toContain('probe?: {'); + expect(diagnosticsResultsPanelSource).toContain("probe().connected ? 'connected' : 'failed'"); expect(diagnosticsModelSource).not.toContain('mcpConnected'); expect(diagnosticsModelSource).not.toContain('mcpToolCount'); expect(diagnosticsModelSource).not.toContain('export interface CommercialFunnelDiagnostic'); diff --git a/frontend-modern/src/components/Settings/diagnosticsModel.ts b/frontend-modern/src/components/Settings/diagnosticsModel.ts index 199b9cea05..b8c61ef4da 100644 --- a/frontend-modern/src/components/Settings/diagnosticsModel.ts +++ b/frontend-modern/src/components/Settings/diagnosticsModel.ts @@ -16,8 +16,18 @@ export interface DiagnosticsPBS { name: string; host: string; connected: boolean; + state?: 'active' | 'paused' | 'unauthorized' | 'unreachable' | 'stale' | 'pending'; + stateReason?: string; + lastSeen?: string; error?: string; details?: Record; + probe?: { + connected: boolean; + error?: string; + errorKind?: string; + troubleshooting?: string; + details?: Record; + }; } export interface SystemDiagnostic { diff --git a/internal/api/config_handlers_connection_test.go b/internal/api/config_handlers_connection_test.go index 1132cb550b..5c7e3b549c 100644 --- a/internal/api/config_handlers_connection_test.go +++ b/internal/api/config_handlers_connection_test.go @@ -6,6 +6,8 @@ import ( "net/http" "net/http/httptest" "os" + "reflect" + "strings" "testing" "github.com/rcourtman/pulse-go-rewrite/internal/config" @@ -125,3 +127,46 @@ func TestHandleTestConnection(t *testing.T) { }) } } + +func TestHandleTestNodePBSIsADirectProbeWithoutChangingRuntimeHealth(t *testing.T) { + var authorization string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + authorization = r.Header.Get("Authorization") + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{}}) + })) + defer server.Close() + + cfg := &config.Config{ + DataPath: t.TempDir(), + PBSInstances: []config.PBSInstance{{ + Name: "backup", + Host: server.URL, + TokenName: "root@pam!pulse", + TokenValue: "rotated-secret", + }}, + } + handler := newTestConfigHandlers(t, cfg) + beforeHealth := handler.defaultMonitor.SchedulerHealth() + beforeStatuses := handler.defaultMonitor.GetConnectionStatuses() + + req := httptest.NewRequest(http.MethodPost, "/api/config/nodes/pbs-0/test", nil) + rec := httptest.NewRecorder() + handler.HandleTestNode(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("saved PBS test status = %d, body=%s", rec.Code, rec.Body.String()) + } + if !strings.Contains(authorization, "rotated-secret") { + t.Fatalf("saved PBS test did not use the current stored token: %q", authorization) + } + afterHealth := handler.defaultMonitor.SchedulerHealth() + if !reflect.DeepEqual(afterHealth.Instances, beforeHealth.Instances) || + !reflect.DeepEqual(afterHealth.Breakers, beforeHealth.Breakers) || + !reflect.DeepEqual(afterHealth.Staleness, beforeHealth.Staleness) { + t.Fatalf("direct PBS test mutated canonical scheduler health:\nbefore=%+v\nafter=%+v", beforeHealth, afterHealth) + } + if !reflect.DeepEqual(handler.defaultMonitor.GetConnectionStatuses(), beforeStatuses) { + t.Fatal("direct PBS test mutated the runtime connection projection") + } +} diff --git a/internal/api/connections_aggregator.go b/internal/api/connections_aggregator.go index a468a066ff..4c8a218e8c 100644 --- a/internal/api/connections_aggregator.go +++ b/internal/api/connections_aggregator.go @@ -1293,7 +1293,11 @@ func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.T return ConnectionStateUnauthorized, lastError.Message, lastSeen, lastError } - if lastSeen == nil && lastError != nil { + // PollStatus.LastError is a current outstanding failure, not historical + // metadata: recordTaskResult clears it on the next success. A prior success + // therefore preserves LastSeen for freshness context but must not make a + // newly failed connection appear active while its breaker is still closed. + if lastError != nil { return ConnectionStateUnreachable, lastError.Message, lastSeen, lastError } diff --git a/internal/api/connections_aggregator_test.go b/internal/api/connections_aggregator_test.go index e64eb66552..4deb469932 100644 --- a/internal/api/connections_aggregator_test.go +++ b/internal/api/connections_aggregator_test.go @@ -5,6 +5,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/internal/monitoring" @@ -1038,6 +1039,86 @@ func TestDeriveConnectionState_FirstPollFailureIsUnreachable(t *testing.T) { } } +func TestDeriveConnectionState_CurrentFailureOverridesRecentSuccess(t *testing.T) { + now := time.Now() + lastSuccess := now.Add(-5 * time.Second) + + for _, test := range []struct { + name string + message string + wantState ConnectionState + }{ + { + name: "authentication rejected", + message: "401 Unauthorized: token invalid", + wantState: ConnectionStateUnauthorized, + }, + { + name: "network timeout", + message: "context deadline exceeded while awaiting response headers", + wantState: ConnectionStateUnreachable, + }, + } { + t.Run(test.name, func(t *testing.T) { + h := healthEntry(&lastSuccess, test.message, "transient", "closed") + state, reason, gotLastSeen, lastError := deriveConnectionState(true, h, now) + if state != test.wantState { + t.Fatalf("state = %q, want %q", state, test.wantState) + } + if reason != test.message || lastError == nil || lastError.Message != test.message { + t.Fatalf("failure detail = reason %q error %+v, want %q", reason, lastError, test.message) + } + if gotLastSeen == nil || !gotLastSeen.Equal(lastSuccess) { + t.Fatalf("lastSeen = %+v, want preserved success %s", gotLastSeen, lastSuccess) + } + }) + } +} + +func TestBuildConnectionsKeepsDistinctPBSSystemsWithSameHostname(t *testing.T) { + now := time.Now() + lastSuccess := now.Add(-10 * time.Second) + got := buildConnections(aggregatorInputs{ + pbsInstances: []config.PBSInstance{ + {Name: "pbs-east", Host: "https://backup.local:8007"}, + {Name: "pbs-west", Host: "https://backup.local:8007"}, + }, + instanceHealth: map[string]monitoring.InstanceHealth{ + "pbs::pbs-east": healthEntry(&lastSuccess, "", "", "closed"), + "pbs::pbs-west": healthEntry(&lastSuccess, "dial tcp: i/o timeout", "transient", "closed"), + }, + now: now, + }) + if len(got) != 2 { + t.Fatalf("connections = %d, want two distinct PBS rows", len(got)) + } + byID := make(map[string]Connection, len(got)) + for _, connection := range got { + byID[connection.ID] = connection + } + if byID["pbs:pbs-east"].State != ConnectionStateActive { + t.Fatalf("pbs-east = %+v, want active", byID["pbs:pbs-east"]) + } + if byID["pbs:pbs-west"].State != ConnectionStateUnreachable { + t.Fatalf("pbs-west = %+v, want unreachable", byID["pbs:pbs-west"]) + } + + alertSnapshots := snapshotConnectionsForAlerts(got) + if len(alertSnapshots) != 2 { + t.Fatalf("alert snapshots = %+v, want both PBS systems", alertSnapshots) + } + alertsByID := make(map[string]alerts.ConnectionSnapshot, len(alertSnapshots)) + for _, snapshot := range alertSnapshots { + alertsByID[snapshot.ID] = snapshot + } + if alertsByID["pbs:pbs-east"].State != alerts.ConnectionStateActive { + t.Fatalf("pbs-east alert state = %+v, want active", alertsByID["pbs:pbs-east"]) + } + if west := alertsByID["pbs:pbs-west"]; west.State != alerts.ConnectionStateUnreachable || west.LastError == nil { + t.Fatalf("pbs-west alert state = %+v, want unreachable with current error", west) + } +} + func TestBuildConnections_UsesHealthLookup(t *testing.T) { now := time.Now() ls := now.Add(-15 * time.Second) diff --git a/internal/api/diagnostics.go b/internal/api/diagnostics.go index 3ebed717ad..c77e646117 100644 --- a/internal/api/diagnostics.go +++ b/internal/api/diagnostics.go @@ -559,10 +559,26 @@ type ClusterInfo struct { // PBSDiagnostic contains diagnostic info for a PBS instance type PBSDiagnostic struct { - ID string `json:"id"` - Name string `json:"name"` - Host string `json:"host"` - AuthMethod string `json:"authMethod"` + ID string `json:"id"` + Name string `json:"name"` + Host string `json:"host"` + AuthMethod string `json:"authMethod"` + Connected bool `json:"connected"` + State ConnectionState `json:"state"` + StateReason string `json:"stateReason,omitempty"` + LastSeen *time.Time `json:"lastSeen,omitempty"` + LastError *ConnectionError `json:"lastError,omitempty"` + Error string `json:"error,omitempty"` + ErrorKind string `json:"errorKind,omitempty"` + Troubleshooting string `json:"troubleshooting,omitempty"` + Details *PBSDetails `json:"details,omitempty"` + Probe PBSProbeDiagnostic `json:"probe"` +} + +// PBSProbeDiagnostic keeps the direct diagnostics request separate from the +// canonical monitor result. A successful one-off probe is useful recovery +// evidence, but it cannot overwrite the last completed runtime poll. +type PBSProbeDiagnostic struct { Connected bool `json:"connected"` Error string `json:"error,omitempty"` ErrorKind string `json:"errorKind,omitempty"` @@ -943,10 +959,32 @@ func applyNodeDiagnosticFailure(diag *NodeDiagnostic, failure diagnosticsConnect } func applyPBSDiagnosticFailure(diag *PBSDiagnostic, failure diagnosticsConnectionFailure) { - diag.Connected = false - diag.Error = failure.message - diag.ErrorKind = failure.kind - diag.Troubleshooting = failure.troubleshooting + diag.Probe.Connected = false + diag.Probe.Error = failure.message + diag.Probe.ErrorKind = failure.kind + diag.Probe.Troubleshooting = failure.troubleshooting +} + +func applyPBSConnectionDiagnostic(diag *PBSDiagnostic, connection Connection) { + if diag == nil { + return + } + diag.State = connection.State + diag.StateReason = connection.StateReason + diag.LastSeen = connection.LastSeen + diag.LastError = connection.LastError + diag.Connected = connection.State == ConnectionStateActive + if diag.Connected { + return + } + + diag.Error = connection.StateReason + if connection.LastError != nil { + failure := classifyDiagnosticsConnectionFailure(errors.New(connection.LastError.Message), connection.StateReason) + diag.Error = failure.message + diag.ErrorKind = failure.kind + diag.Troubleshooting = failure.troubleshooting + } } func (r *Router) computeDiagnostics(ctx context.Context) DiagnosticsInfo { @@ -1058,7 +1096,21 @@ func (r *Router) computeDiagnostics(ctx context.Context) DiagnosticsInfo { diag.Nodes = append(diag.Nodes, nodeDiag) } - // Test PBS instances + pbsConnections := make(map[string]Connection) + for _, connection := range buildConnections(buildAggregatorInputsWithRuntimeSources( + ctx, + r.config, + r.persistence, + r.monitor, + aggregatorRuntimeSources{}, + )) { + if connection.Type == ConnectionTypePBS { + pbsConnections[connection.Name] = connection + } + } + + // Test PBS instances. Connected/state use the canonical scheduler ledger; + // Probe records this request's direct check as separate evidence. for _, pbsNode := range r.config.PBSInstances { pbsDiag := PBSDiagnostic{ ID: pbsNode.Name, @@ -1066,11 +1118,9 @@ func (r *Router) computeDiagnostics(ctx context.Context) DiagnosticsInfo { Host: pbsNode.Host, AuthMethod: diagnosticsAuthMethod(pbsNode.User, pbsNode.Password, pbsNode.TokenName, pbsNode.TokenValue), } - pbsKeyName := pbsNode.Name - if strings.TrimSpace(pbsKeyName) == "" { - pbsKeyName = pbsNode.Host + if connection, ok := pbsConnections[pbsNode.Name]; ok { + applyPBSConnectionDiagnostic(&pbsDiag, connection) } - monitorConnected, hasMonitorStatus := diagnosticsMonitorConnectionStatus(r.monitor, "pbs-"+pbsKeyName) if pbsDiag.AuthMethod == "none" { applyPBSDiagnosticFailure(&pbsDiag, missingStoredSecretDiagnosticsFailure()) diag.PBS = append(diag.PBS, pbsDiag) @@ -1092,16 +1142,15 @@ func (r *Router) computeDiagnostics(ctx context.Context) DiagnosticsInfo { failure := classifyDiagnosticsConnectionFailure(err, "Failed to initialize connection") applyPBSDiagnosticFailure(&pbsDiag, failure) log.Error().Err(err).Str("pbs", pbsNode.Name).Msg("Diagnostics: PBS client init failed") - pbsDiag.Connected, pbsDiag.Error = mergeDiagnosticsConnection(pbsDiag.Connected, pbsDiag.Error, monitorConnected, hasMonitorStatus) } else { if version, err := client.GetVersion(ctx); err != nil { failure := classifyDiagnosticsConnectionFailure(err, "Connection established but version check failed") applyPBSDiagnosticFailure(&pbsDiag, failure) log.Error().Err(err).Str("pbs", pbsNode.Name).Msg("Diagnostics: PBS version check failed") - pbsDiag.Connected, pbsDiag.Error = mergeDiagnosticsConnection(pbsDiag.Connected, pbsDiag.Error, monitorConnected, hasMonitorStatus) } else { - pbsDiag.Connected = true + pbsDiag.Probe.Connected = true pbsDiag.Details = &PBSDetails{Version: version.Version} + pbsDiag.Probe.Details = pbsDiag.Details } } diff --git a/internal/api/diagnostics_additional_test.go b/internal/api/diagnostics_additional_test.go index 070e4548b1..e5fead6201 100644 --- a/internal/api/diagnostics_additional_test.go +++ b/internal/api/diagnostics_additional_test.go @@ -134,13 +134,13 @@ func TestComputeDiagnosticsClassifiesMissingNodeSecrets(t *testing.T) { if got := diag.PBS[0].AuthMethod; got != "none" { t.Fatalf("PBS auth method = %q, want none", got) } - if got := diag.PBS[0].ErrorKind; got != "no_stored_secret" { + if got := diag.PBS[0].Probe.ErrorKind; got != "no_stored_secret" { t.Fatalf("PBS error kind = %q, want no_stored_secret", got) } - if !strings.Contains(diag.PBS[0].Error, "no stored credentials") { - t.Fatalf("PBS error should describe missing credentials, got %q", diag.PBS[0].Error) + if !strings.Contains(diag.PBS[0].Probe.Error, "no stored credentials") { + t.Fatalf("PBS probe error should describe missing credentials, got %q", diag.PBS[0].Probe.Error) } - if strings.TrimSpace(diag.PBS[0].Troubleshooting) == "" { + if strings.TrimSpace(diag.PBS[0].Probe.Troubleshooting) == "" { t.Fatal("expected PBS troubleshooting guidance") } } diff --git a/internal/api/diagnostics_test.go b/internal/api/diagnostics_test.go index 33bb2d90db..8430e3f31d 100644 --- a/internal/api/diagnostics_test.go +++ b/internal/api/diagnostics_test.go @@ -7,6 +7,8 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/monitoring" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) @@ -800,6 +802,56 @@ func TestMergeDiagnosticsConnection(t *testing.T) { } } +func TestPBSDiagnosticsKeepsCanonicalPollStateSeparateFromLiveProbe(t *testing.T) { + now := time.Now() + lastSuccess := now.Add(-time.Minute) + connections := buildConnections(aggregatorInputs{ + pbsInstances: []config.PBSInstance{{ + Name: "pbs-primary", + Host: "https://backup.local:8007", + }}, + instanceHealth: map[string]monitoring.InstanceHealth{ + "pbs::pbs-primary": { + PollStatus: monitoring.InstancePollStatus{ + LastSuccess: &lastSuccess, + LastError: &monitoring.ErrorDetail{ + At: now, + Message: "connection timed out", + }, + }, + }, + }, + now: now, + }) + if len(connections) != 1 { + t.Fatalf("connections = %+v, want one PBS connection", connections) + } + + diagnostic := PBSDiagnostic{ + ID: "pbs-primary", + Name: "pbs-primary", + Host: "https://backup.local:8007", + Probe: PBSProbeDiagnostic{ + Connected: true, + Details: &PBSDetails{Version: "3.4.2"}, + }, + } + applyPBSConnectionDiagnostic(&diagnostic, connections[0]) + + if diagnostic.Connected { + t.Fatal("canonical diagnostics state became connected from a successful direct probe") + } + if diagnostic.State != ConnectionStateUnreachable { + t.Fatalf("state = %q, want unreachable", diagnostic.State) + } + if !diagnostic.Probe.Connected { + t.Fatal("successful direct probe evidence was lost") + } + if diagnostic.LastSeen == nil || !diagnostic.LastSeen.Equal(lastSuccess) { + t.Fatalf("last seen = %v, want %s", diagnostic.LastSeen, lastSuccess) + } +} + func TestClassifyDiagnosticsConnectionFailure(t *testing.T) { cases := []struct { name string diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index 16c96ff127..f57874ee54 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -90,6 +90,22 @@ func TestNoGetStateResourceArrayRegression(t *testing.T) { } } +func TestPBSClientConstructionCannotPublishConnectedHealth(t *testing.T) { + for _, name := range []string{"monitor_client_init.go", "monitor_client_reconnect.go"} { + data, err := os.ReadFile(name) + if err != nil { + t.Fatalf("read %s: %v", name, err) + } + source := string(data) + if strings.Contains(source, "setProviderConnectionHealth(InstanceTypePBS") { + t.Fatalf("%s must wait for a completed PBS poll before publishing connection health", name) + } + if !strings.Contains(source, "awaiting") { + t.Fatalf("%s must make the first-poll boundary explicit", name) + } + } +} + func TestMultiTenantMonitorListOrganizationIDsFallsBackToDefault(t *testing.T) { got, err := (&MultiTenantMonitor{}).ListOrganizationIDs() if err != nil { diff --git a/internal/monitoring/monitor_client_init.go b/internal/monitoring/monitor_client_init.go index fe0d2b941d..733180578d 100644 --- a/internal/monitoring/monitor_client_init.go +++ b/internal/monitoring/monitor_client_init.go @@ -4,6 +4,7 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" "github.com/rcourtman/pulse-go-rewrite/pkg/pmg" @@ -106,14 +107,24 @@ func (m *Monitor) initPBSClients(cfg *config.Config) { Bool("hasPassword", pbsInst.Password != ""). Bool("hasToken", pbsInst.TokenValue != ""). Msg("Failed to create PBS client - node will show as disconnected") - // Set initial connection health to false for this node - m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, false) + m.recordTaskResult(InstanceTypePBS, pbsInst.Name, monErr) + if m.stalenessTracker != nil { + m.stalenessTracker.UpdateError(InstanceTypePBS, pbsInst.Name) + } + m.publishPBSConnectionOutcome(models.PBSInstance{ + ID: "pbs-" + pbsInst.Name, + Name: pbsInst.Name, + Host: pbsInst.Host, + GuestURL: pbsInst.GuestURL, + Version: "unknown", + LastSeen: time.Now(), + }, monErr) continue } m.pbsClients[pbsInst.Name] = client - log.Info().Str("instance", pbsInst.Name).Msg("PBS client created successfully") - // Set initial connection health to true - m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, true) + log.Info(). + Str("instance", pbsInst.Name). + Msg("PBS client created; awaiting first connectivity poll") } } diff --git a/internal/monitoring/monitor_client_reconnect.go b/internal/monitoring/monitor_client_reconnect.go index e3cd2c0be0..d2f9f509b6 100644 --- a/internal/monitoring/monitor_client_reconnect.go +++ b/internal/monitoring/monitor_client_reconnect.go @@ -5,6 +5,8 @@ import ( "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + monitorerrors "github.com/rcourtman/pulse-go-rewrite/internal/monitoring/errors" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" "github.com/rs/zerolog/log" @@ -154,21 +156,33 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) { clientConfig.Timeout = 60 * time.Second client, err := pbs.NewClient(clientConfig) if err != nil { + monErr := monitorerrors.WrapConnectionError("recreate_pbs_client", pbsInst.Name, err) log.Warn(). - Err(err). + Err(monErr). Str("instance", pbsInst.Name). Msg("Failed to reconnect PBS client, will retry") + m.recordTaskResult(InstanceTypePBS, pbsInst.Name, monErr) + if m.stalenessTracker != nil { + m.stalenessTracker.UpdateError(InstanceTypePBS, pbsInst.Name) + } + m.publishPBSConnectionOutcome(models.PBSInstance{ + ID: "pbs-" + pbsInst.Name, + Name: pbsInst.Name, + Host: pbsInst.Host, + GuestURL: pbsInst.GuestURL, + Version: "unknown", + LastSeen: time.Now(), + }, monErr) continue } m.mu.Lock() m.pbsClients[pbsInst.Name] = client m.mu.Unlock() - m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, true) log.Info(). Str("instance", pbsInst.Name). - Msg("Successfully reconnected PBS client") + Msg("PBS client recreated; awaiting connectivity poll") } } } diff --git a/internal/monitoring/monitor_full_coverage_test.go b/internal/monitoring/monitor_full_coverage_test.go index a10ddc664c..8cc83e00ef 100644 --- a/internal/monitoring/monitor_full_coverage_test.go +++ b/internal/monitoring/monitor_full_coverage_test.go @@ -139,6 +139,7 @@ func TestPollPBSInstance(t *testing.T) { }, state: models.NewState(), stalenessTracker: NewStalenessTracker(nil), // Pass nil or mock PollMetrics + pollStatusMap: make(map[string]*pollStatus), nodePendingUpdatesCache: make(map[string]pendingUpdatesCache), } diff --git a/internal/monitoring/monitor_pbs_coverage_test.go b/internal/monitoring/monitor_pbs_coverage_test.go index bb0b9d9bf1..453c585f64 100644 --- a/internal/monitoring/monitor_pbs_coverage_test.go +++ b/internal/monitoring/monitor_pbs_coverage_test.go @@ -63,6 +63,12 @@ func TestMonitor_PollPBSInstance_AuthFailure(t *testing.T) { if snapshot.ConnectionHealth["pbs-pbs-auth-fail"] { t.Error("Expected connection health to be false") } + if len(snapshot.PBSInstances) != 1 { + t.Fatalf("expected failed PBS poll to refresh one dashboard projection, got %+v", snapshot.PBSInstances) + } + if instance := snapshot.PBSInstances[0]; instance.Status != "offline" || instance.ConnectionHealth != "error" { + t.Fatalf("failed PBS dashboard projection = %+v, want offline/error", instance) + } // Regression: pollStatusMap must record the failure. A defer-arg bug // previously captured pollErr at register-time (always nil), so failed diff --git a/internal/monitoring/monitor_pbs_health_authority_test.go b/internal/monitoring/monitor_pbs_health_authority_test.go new file mode 100644 index 0000000000..a7f20894af --- /dev/null +++ b/internal/monitoring/monitor_pbs_health_authority_test.go @@ -0,0 +1,299 @@ +package monitoring + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/pkg/pbs" +) + +type pbsHealthTestMode int32 + +const ( + pbsHealthTestSuccess pbsHealthTestMode = iota + pbsHealthTestAuthFailure + pbsHealthTestTimeout + pbsHealthTestPartialData +) + +type pbsHealthTestServer struct { + server *httptest.Server + mode atomic.Int32 + requiredToken atomic.Value +} + +func newPBSHealthTestServer(t *testing.T) *pbsHealthTestServer { + t.Helper() + + fixture := &pbsHealthTestServer{} + fixture.mode.Store(int32(pbsHealthTestSuccess)) + fixture.requiredToken.Store("") + fixture.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if requiredToken := fixture.requiredToken.Load().(string); requiredToken != "" && + !strings.Contains(r.Header.Get("Authorization"), requiredToken) { + http.Error(w, "authentication failed: 401 Unauthorized", http.StatusUnauthorized) + return + } + mode := pbsHealthTestMode(fixture.mode.Load()) + switch mode { + case pbsHealthTestAuthFailure: + http.Error(w, "authentication failed: 401 Unauthorized", http.StatusUnauthorized) + return + case pbsHealthTestTimeout: + <-r.Context().Done() + return + } + + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/api2/json/version": + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": map[string]any{"version": "3.4.2"}, + }) + case "/api2/json/nodes/localhost/status": + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": map[string]any{ + "cpu": 0.15, + "memory": map[string]any{"used": 512, "total": 1024}, + "uptime": 120, + }, + }) + case "/api2/json/admin/datastore": + if mode == pbsHealthTestPartialData { + http.Error(w, "datastore inventory temporarily unavailable", http.StatusServiceUnavailable) + return + } + _ = json.NewEncoder(w).Encode(map[string]any{ + "data": []map[string]any{{ + "store": "backups", + "total": 1000, + "used": 250, + "avail": 750, + }}, + }) + case "/api2/json/admin/datastore/backups/namespace": + _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{}}) + default: + _ = json.NewEncoder(w).Encode(map[string]any{"data": []any{}}) + } + })) + t.Cleanup(fixture.server.Close) + return fixture +} + +func (s *pbsHealthTestServer) setMode(mode pbsHealthTestMode) { + s.mode.Store(int32(mode)) +} + +func (s *pbsHealthTestServer) requireToken(token string) { + s.requiredToken.Store(token) +} + +func newPBSHealthAuthorityMonitor(instances []config.PBSInstance) *Monitor { + return &Monitor{ + config: &config.Config{PBSInstances: instances}, + state: models.NewState(), + pbsClients: make(map[string]*pbs.Client), + authFailures: make(map[string]int), + lastAuthAttempt: make(map[string]time.Time), + pollStatusMap: make(map[string]*pollStatus), + circuitBreakers: make(map[string]*circuitBreaker), + failureCounts: make(map[string]int), + lastOutcome: make(map[string]taskOutcome), + stalenessTracker: NewStalenessTracker(nil), + } +} + +func newPBSHealthTestClient(t *testing.T, host string) *pbs.Client { + return newPBSHealthTestClientWithToken(t, host, "secret") +} + +func newPBSHealthTestClientWithToken(t *testing.T, host, token string) *pbs.Client { + t.Helper() + client, err := pbs.NewClient(pbs.ClientConfig{ + Host: host, + TokenName: "root@pam!pulse", + TokenValue: token, + Timeout: 75 * time.Millisecond, + }) + if err != nil { + t.Fatalf("new PBS client: %v", err) + } + return client +} + +func pbsInstanceByName(t *testing.T, snapshot models.StateSnapshot, name string) models.PBSInstance { + t.Helper() + for _, instance := range snapshot.PBSInstances { + if instance.Name == name { + return instance + } + } + t.Fatalf("PBS instance %q missing from snapshot: %+v", name, snapshot.PBSInstances) + return models.PBSInstance{} +} + +func assertPBSConnectionProjection( + t *testing.T, + monitor *Monitor, + name string, + wantConnected bool, + wantStatus string, +) { + t.Helper() + snapshot := monitor.state.GetSnapshot() + instance := pbsInstanceByName(t, snapshot, name) + if got := snapshot.ConnectionHealth["pbs-"+name]; got != wantConnected { + t.Fatalf("%s connection-health map = %v, want %v", name, got, wantConnected) + } + if got := instance.Status; got != wantStatus { + t.Fatalf("%s dashboard status = %q, want %q", name, got, wantStatus) + } + wantResourceHealth := "healthy" + if !wantConnected { + wantResourceHealth = "error" + } + if got := instance.ConnectionHealth; got != wantResourceHealth { + t.Fatalf("%s dashboard connection health = %q, want %q", name, got, wantResourceHealth) + } +} + +func TestPollPBSInstancesKeepsAllHealthProjectionsOnOneOutcome(t *testing.T) { + first := newPBSHealthTestServer(t) + second := newPBSHealthTestServer(t) + instances := []config.PBSInstance{ + {Name: "pbs-primary", Host: first.server.URL, MonitorDatastores: true}, + {Name: "pbs-secondary", Host: second.server.URL, MonitorDatastores: true}, + } + monitor := newPBSHealthAuthorityMonitor(instances) + clients := map[string]*pbs.Client{ + "pbs-primary": newPBSHealthTestClient(t, first.server.URL), + "pbs-secondary": newPBSHealthTestClient(t, second.server.URL), + } + + for name, client := range clients { + monitor.pollPBSInstance(context.Background(), name, client) + assertPBSConnectionProjection(t, monitor, name, true, "online") + } + primarySuccess := monitor.pollStatusMap["pbs::pbs-primary"].LastSuccess + secondarySuccess := monitor.pollStatusMap["pbs::pbs-secondary"].LastSuccess + if primarySuccess.IsZero() || secondarySuccess.IsZero() { + t.Fatalf("initial successes were not recorded: primary=%s secondary=%s", primarySuccess, secondarySuccess) + } + + first.setMode(pbsHealthTestAuthFailure) + second.setMode(pbsHealthTestTimeout) + monitor.pollPBSInstance(context.Background(), "pbs-primary", clients["pbs-primary"]) + monitor.pollPBSInstance(context.Background(), "pbs-secondary", clients["pbs-secondary"]) + + for name, previousSuccess := range map[string]time.Time{ + "pbs-primary": primarySuccess, + "pbs-secondary": secondarySuccess, + } { + status := monitor.pollStatusMap["pbs::"+name] + if status == nil || status.LastErrorMessage == "" || status.ConsecutiveFailures != 1 { + t.Fatalf("%s failure ledger = %+v, want one current failure", name, status) + } + if !status.LastSuccess.Equal(previousSuccess) { + t.Fatalf("%s last success changed on failure: got %s want %s", name, status.LastSuccess, previousSuccess) + } + assertPBSConnectionProjection(t, monitor, name, false, "offline") + } + + first.setMode(pbsHealthTestSuccess) + first.requireToken("rotated-secret") + monitor.pollPBSInstance(context.Background(), "pbs-primary", clients["pbs-primary"]) + assertPBSConnectionProjection(t, monitor, "pbs-primary", false, "offline") + + clients["pbs-primary"] = newPBSHealthTestClientWithToken(t, first.server.URL, "rotated-secret") + monitor.pollPBSInstance(context.Background(), "pbs-primary", clients["pbs-primary"]) + assertPBSConnectionProjection(t, monitor, "pbs-primary", true, "online") + + replacement := newPBSHealthTestServer(t) + monitor.config.PBSInstances[1].Host = replacement.server.URL + clients["pbs-secondary"] = newPBSHealthTestClient(t, replacement.server.URL) + time.Sleep(time.Millisecond) + monitor.pollPBSInstance(context.Background(), "pbs-secondary", clients["pbs-secondary"]) + if got := pbsInstanceByName(t, monitor.state.GetSnapshot(), "pbs-secondary").Host; got != replacement.server.URL { + t.Fatalf("reconfigured PBS host = %q, want %q", got, replacement.server.URL) + } + + for name := range clients { + status := monitor.pollStatusMap["pbs::"+name] + if status == nil || !status.LastErrorAt.IsZero() || status.LastErrorMessage != "" || status.ConsecutiveFailures != 0 { + t.Fatalf("%s recovery ledger = %+v, want cleared current failure", name, status) + } + assertPBSConnectionProjection(t, monitor, name, true, "online") + } + + restarted := newPBSHealthAuthorityMonitor(monitor.config.PBSInstances) + for name, client := range clients { + restarted.pollPBSInstance(context.Background(), name, client) + assertPBSConnectionProjection(t, restarted, name, true, "online") + } +} + +func TestPollPBSInstanceKeepsPartialDataSeparateFromConnectivity(t *testing.T) { + fixture := newPBSHealthTestServer(t) + fixture.setMode(pbsHealthTestPartialData) + instance := config.PBSInstance{ + Name: "pbs-partial", + Host: fixture.server.URL, + MonitorDatastores: true, + } + monitor := newPBSHealthAuthorityMonitor([]config.PBSInstance{instance}) + + monitor.pollPBSInstance(context.Background(), instance.Name, newPBSHealthTestClient(t, instance.Host)) + + assertPBSConnectionProjection(t, monitor, instance.Name, true, "online") + status := monitor.pollStatusMap["pbs::"+instance.Name] + if status == nil || status.LastSuccess.IsZero() || status.LastErrorMessage != "" { + t.Fatalf("partial collection must retain successful connectivity, got %+v", status) + } + published := pbsInstanceByName(t, monitor.state.GetSnapshot(), instance.Name) + if len(published.Datastores) != 0 { + t.Fatalf("partial datastore inventory = %+v, want no fabricated rows", published.Datastores) + } +} + +func TestInitPBSClientsDoesNotTreatClientConstructionAsConnectivity(t *testing.T) { + instances := []config.PBSInstance{ + { + Name: "pbs-awaiting-poll", + Host: "https://backup.local:8007", + TokenName: "root@pam!pulse", + TokenValue: "secret", + }, + { + Name: "pbs-invalid-url", + Host: "://not-a-url", + TokenName: "root@pam!pulse", + TokenValue: "secret", + }, + } + monitor := newPBSHealthAuthorityMonitor(instances) + + monitor.initPBSClients(monitor.config) + + snapshot := monitor.state.GetSnapshot() + if _, ok := snapshot.ConnectionHealth["pbs-pbs-awaiting-poll"]; ok { + t.Fatal("successful client construction fabricated a connected health result") + } + if _, ok := monitor.pollStatusMap["pbs::pbs-awaiting-poll"]; ok { + t.Fatal("successful client construction fabricated a completed poll") + } + + status := monitor.pollStatusMap["pbs::pbs-invalid-url"] + if status == nil || status.LastErrorMessage == "" || status.ConsecutiveFailures != 1 { + t.Fatalf("invalid URL initialization ledger = %+v, want one current failure", status) + } + assertPBSConnectionProjection(t, monitor, "pbs-invalid-url", false, "offline") +} diff --git a/internal/monitoring/monitor_pbs_pmg.go b/internal/monitoring/monitor_pbs_pmg.go index 105ecae6da..c1a74a7c82 100644 --- a/internal/monitoring/monitor_pbs_pmg.go +++ b/internal/monitoring/monitor_pbs_pmg.go @@ -167,15 +167,37 @@ func pbsJobHealthPosture(fact pbs.JobHealthEvidence, freshness models.PBSJobHeal // pollPBSInstance polls a single PBS instance func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, client *pbs.Client) { - defer recoverFromPanic(fmt.Sprintf("pollPBSInstance-%s", instanceName)) - start := time.Now() debugEnabled := logging.IsLevelEnabled(zerolog.DebugLevel) var pollErr error + var pbsInst models.PBSInstance + publishResult := false if m.pollMetrics != nil { m.pollMetrics.IncInFlight("pbs") - defer m.pollMetrics.DecInFlight("pbs") - defer func() { + } + defer func() { + if recovered := recover(); recovered != nil { + pollErr = fmt.Errorf("panic while polling PBS instance %q: %v", instanceName, recovered) + log.Error(). + Str("goroutine", fmt.Sprintf("pollPBSInstance-%s", instanceName)). + Interface("panic", recovered). + Stack(). + Msg("Recovered from panic in monitoring goroutine") + } + + m.recordTaskResult(InstanceTypePBS, instanceName, pollErr) + if m.stalenessTracker != nil { + if pollErr == nil { + m.stalenessTracker.UpdateSuccess(InstanceTypePBS, instanceName, nil) + } else { + m.stalenessTracker.UpdateError(InstanceTypePBS, instanceName) + } + } + if publishResult { + m.publishPBSConnectionOutcome(pbsInst, pollErr) + } + if m.pollMetrics != nil { + m.pollMetrics.DecInFlight("pbs") m.pollMetrics.RecordResult(PollResult{ InstanceName: instanceName, InstanceType: "pbs", @@ -184,35 +206,8 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie StartTime: start, EndTime: time.Now(), }) - }() - } - if m.stalenessTracker != nil { - defer func() { - if pollErr == nil { - m.stalenessTracker.UpdateSuccess(InstanceTypePBS, instanceName, nil) - } else { - m.stalenessTracker.UpdateError(InstanceTypePBS, instanceName) - } - }() - } - defer func() { - m.recordTaskResult(InstanceTypePBS, instanceName, pollErr) - }() - - // Check if context is cancelled - select { - case <-ctx.Done(): - pollErr = ctx.Err() - if debugEnabled { - log.Debug().Str("instance", instanceName).Msg("polling cancelled") } - return - default: - } - - if debugEnabled { - log.Debug().Str("instance", instanceName).Msg("polling PBS instance") - } + }() // Get instance config var instanceCfg *config.PBSInstance @@ -229,6 +224,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } } if instanceCfg == nil { + pollErr = fmt.Errorf("PBS instance config %q not found", instanceName) log.Error().Str("instance", instanceName).Msg("PBS instance config not found") return } @@ -240,7 +236,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } // Initialize PBS instance with default values - pbsInst := models.PBSInstance{ + pbsInst = models.PBSInstance{ ID: "pbs-" + instanceName, Name: instanceName, Host: instanceCfg.Host, @@ -250,6 +246,23 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie ConnectionHealth: "unhealthy", LastSeen: time.Now(), } + publishResult = true + + // Check if context is cancelled after the configured instance projection is + // available, so cancellation still publishes the same failed poll outcome. + select { + case <-ctx.Done(): + pollErr = ctx.Err() + if debugEnabled { + log.Debug().Str("instance", instanceName).Msg("polling cancelled") + } + return + default: + } + + if debugEnabled { + log.Debug().Str("instance", instanceName).Msg("polling PBS instance") + } // Try to get version first version, versionErr := client.GetVersion(ctx) @@ -258,7 +271,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie pbsInst.Version = version.Version pbsInst.ConnectionHealth = "healthy" m.resetAuthFailures(instanceName, "pbs") - m.setProviderConnectionHealth(InstanceTypePBS, instanceName, true) if debugEnabled { log.Debug(). @@ -281,7 +293,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie pbsInst.Version = "connected" pbsInst.ConnectionHealth = "healthy" m.resetAuthFailures(instanceName, "pbs") - m.setProviderConnectionHealth(InstanceTypePBS, instanceName, true) log.Info(). Str("instance", instanceName). @@ -289,15 +300,18 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } else { pbsInst.Status = "offline" pbsInst.ConnectionHealth = "error" - monErr := errors.WrapConnectionError("get_pbs_version", instanceName, versionErr) + connectionErr := versionErr + if errors.IsAuthError(datastoreErr) { + connectionErr = datastoreErr + } + monErr := errors.WrapConnectionError("get_pbs_version", instanceName, connectionErr) pollErr = monErr log.Error().Err(monErr).Str("instance", instanceName).Msg("failed to connect to PBS") - m.setProviderConnectionHealth(InstanceTypePBS, instanceName, false) if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) { m.recordAuthFailure(instanceName, "pbs") - return } + return } } @@ -442,14 +456,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie pbsInst.JobHealthEvidence = pbsJobHealthEvidenceFromFacts(jobFacts, time.Now()) } - // Update state and run alerts - m.state.UpdatePBSInstance(pbsInst) - log.Info(). - Str("instance", instanceName). - Str("id", pbsInst.ID). - Int("datastores", len(pbsInst.Datastores)). - Msg("PBS instance updated in state") - // Convert PBS datastores to Storage entries for unified storage view if len(pbsInst.Datastores) > 0 && instanceCfg.MonitorDatastores { var pbsStorages []models.Storage @@ -486,10 +492,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie Msg("Added PBS datastores to unified storage view") } - if m.alertManager != nil { - m.alertManager.CheckPBS(pbsInst) - } - // Poll backups if enabled if instanceCfg.MonitorBackups { if len(pbsInst.Datastores) == 0 { @@ -580,6 +582,34 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie } } +// publishPBSConnectionOutcome projects the final connectivity result into every +// legacy PBS health consumer. Inventory collection may be partial, but the +// connection map, dashboard resource, and legacy alert evaluator must all use +// the same poll error that is recorded in the scheduler ledger. +func (m *Monitor) publishPBSConnectionOutcome(instance models.PBSInstance, pollErr error) { + connected := pollErr == nil + if connected { + instance.Status = "online" + instance.ConnectionHealth = "healthy" + } else { + instance.Status = "offline" + instance.ConnectionHealth = "error" + } + + m.setProviderConnectionHealth(InstanceTypePBS, instance.Name, connected) + m.state.UpdatePBSInstance(instance) + log.Info(). + Str("instance", instance.Name). + Str("id", instance.ID). + Bool("connected", connected). + Int("datastores", len(instance.Datastores)). + Msg("PBS instance updated in state") + + if m.alertManager != nil { + m.alertManager.CheckPBS(instance) + } +} + // pollPMGInstance polls a single Proxmox Mail Gateway instance func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) { defer recoverFromPanic(fmt.Sprintf("pollPMGInstance-%s", instanceName))