Fix PBS connection health authority

This commit is contained in:
rcourtman
2026-07-24 09:45:00 +01:00
parent 6e93cb3b5c
commit 4815a5c80d
24 changed files with 944 additions and 90 deletions
@@ -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.
+27 -1
View File
@@ -5742,6 +5742,12 @@
"kind": "file", "kind": "file",
"evidence_tier": "managed-runtime-exercise" "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", "repo": "pulse",
"path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-proxmox-cluster-address-authority-2026-07-23.md", "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", "kind": "file",
"evidence_tier": "managed-runtime-exercise" "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", "repo": "pulse",
"path": "docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-proxmox-cluster-address-authority-2026-07-23.md", "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": [], "open_decisions": [],
"source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md", "source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md",
"resolved_decisions": [ "resolved_decisions": [
@@ -2101,6 +2101,15 @@ Agent` secondary handoff against the live setup wizard instead of relying
## Current State ## 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 ### Docker and Podman report sizes share one exact-byte contract
The Unified Agent now measures the encoded gzip body and decoded report JSON The Unified Agent now measures the encoded gzip body and decoded report JSON
@@ -5916,6 +5916,17 @@ machine-readable `errorKind` values with operator guidance in
certificate mismatch, refused connections, unreachable networks, DNS failures, certificate mismatch, refused connections, unreachable networks, DNS failures,
timeouts, and unknown connection failures must remain distinguishable in timeouts, and unknown connection failures must remain distinguishable in
support bundles instead of collapsing into a generic connection failure string. 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. That same diagnostics boundary now explicitly excludes maintainer analytics.
`internal/api/diagnostics.go` must not serialize commercial funnel, sales `internal/api/diagnostics.go` must not serialize commercial funnel, sales
funnel, pricing/checkout conversion, or infrastructure onboarding telemetry in funnel, pricing/checkout conversion, or infrastructure onboarding telemetry in
@@ -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 customer-facing diagnostics copy. The shell must not re-accumulate inline API
calls, export-download plumbing, diagnostics-card composition, or diagnostics calls, export-download plumbing, diagnostics-card composition, or diagnostics
surface copy. 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 That same diagnostics owner split also keeps maintainer analytics out of the
customer diagnostics surface. `DiagnosticsResultsPanel.tsx`, customer diagnostics surface. `DiagnosticsResultsPanel.tsx`,
`diagnosticsModel.ts`, and the diagnostics export path must not render or `diagnosticsModel.ts`, and the diagnostics export path must not render or
@@ -644,6 +644,21 @@ node-local Agent evidence.
## Current State ## 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 ### Host snapshots carry integration provenance; doctor copy is user-facing
`models.Host.IntegrationSource` mirrors the unified fabric's `models.Host.IntegrationSource` mirrors the unified fabric's
@@ -1950,6 +1950,18 @@ registered-agent counts does not alter any Recovery Assurance domain:
recovery evidence still comes exclusively from its own deterministic recovery evidence still comes exclusively from its own deterministic
verification paths. 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 actions retain execution-time readiness checks
Recovery-oriented capabilities admitted to canonical Actions are subject to Recovery-oriented capabilities admitted to canonical Actions are subject to
@@ -172,16 +172,30 @@ export const DiagnosticsResultsPanel: Component<DiagnosticsResultsPanelProps> =
<div class="space-y-1"> <div class="space-y-1">
<For each={props.diagnosticsData?.pbs || []}> <For each={props.diagnosticsData?.pbs || []}>
{(pbs) => ( {(pbs) => (
<div class="flex items-center justify-between border-b border-border-subtle py-1 last:border-0"> <div class="border-b border-border-subtle py-1 last:border-0">
<span class="max-w-[120px] truncate" title={pbs.host}> <div class="flex items-center justify-between">
{pbs.name} <span class="max-w-[120px] truncate" title={pbs.host}>
</span> {pbs.name}
<StatusIndicatorBadge </span>
status={pbs.connected ? 'online' : 'offline'} <StatusIndicatorBadge
dot status={pbs.connected ? 'online' : 'offline'}
uppercase dot
size="xs" uppercase
/> size="xs"
/>
</div>
<Show
when={
pbs.probe && pbs.probe.connected !== pbs.connected ? pbs.probe : undefined
}
>
{(probe) => (
<div class="mt-0.5 text-xs text-muted">
Monitor: {pbs.state || (pbs.connected ? 'active' : 'unreachable')}; live
check: {probe().connected ? 'connected' : 'failed'}
</div>
)}
</Show>
</div> </div>
)} )}
</For> </For>
@@ -244,4 +244,54 @@ describe('DiagnosticsResultsPanel', () => {
); );
expect(screen.queryByText('Container Runtime Agents')).not.toBeInTheDocument(); 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(() => (
<Router>
<Route
path="/"
component={() => (
<DiagnosticsResultsPanel
diagnosticsData={diagnosticsData}
loading={false}
onRunDiagnostics={() => {}}
/>
)}
/>
</Router>
));
expect(screen.getByText('pbs-primary')).toBeInTheDocument();
expect(screen.getByText('Monitor: unreachable; live check: connected')).toBeInTheDocument();
});
}); });
@@ -1700,6 +1700,8 @@ describe('settings architecture guardrails', () => {
expect(diagnosticsModelSource).toContain('stripInternalAnalyticsDiagnosticsFields'); expect(diagnosticsModelSource).toContain('stripInternalAnalyticsDiagnosticsFields');
expect(diagnosticsModelSource).toContain('assistantRuntimeConnected'); expect(diagnosticsModelSource).toContain('assistantRuntimeConnected');
expect(diagnosticsModelSource).toContain('probe?: {');
expect(diagnosticsResultsPanelSource).toContain("probe().connected ? 'connected' : 'failed'");
expect(diagnosticsModelSource).not.toContain('mcpConnected'); expect(diagnosticsModelSource).not.toContain('mcpConnected');
expect(diagnosticsModelSource).not.toContain('mcpToolCount'); expect(diagnosticsModelSource).not.toContain('mcpToolCount');
expect(diagnosticsModelSource).not.toContain('export interface CommercialFunnelDiagnostic'); expect(diagnosticsModelSource).not.toContain('export interface CommercialFunnelDiagnostic');
@@ -16,8 +16,18 @@ export interface DiagnosticsPBS {
name: string; name: string;
host: string; host: string;
connected: boolean; connected: boolean;
state?: 'active' | 'paused' | 'unauthorized' | 'unreachable' | 'stale' | 'pending';
stateReason?: string;
lastSeen?: string;
error?: string; error?: string;
details?: Record<string, unknown>; details?: Record<string, unknown>;
probe?: {
connected: boolean;
error?: string;
errorKind?: string;
troubleshooting?: string;
details?: Record<string, unknown>;
};
} }
export interface SystemDiagnostic { export interface SystemDiagnostic {
@@ -6,6 +6,8 @@ import (
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"os" "os"
"reflect"
"strings"
"testing" "testing"
"github.com/rcourtman/pulse-go-rewrite/internal/config" "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")
}
}
+5 -1
View File
@@ -1293,7 +1293,11 @@ func deriveConnectionState(enabled bool, h monitoring.InstanceHealth, now time.T
return ConnectionStateUnauthorized, lastError.Message, lastSeen, lastError 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 return ConnectionStateUnreachable, lastError.Message, lastSeen, lastError
} }
@@ -5,6 +5,7 @@ import (
"testing" "testing"
"time" "time"
"github.com/rcourtman/pulse-go-rewrite/internal/alerts"
"github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/config"
"github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/models"
"github.com/rcourtman/pulse-go-rewrite/internal/monitoring" "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) { func TestBuildConnections_UsesHealthLookup(t *testing.T) {
now := time.Now() now := time.Now()
ls := now.Add(-15 * time.Second) ls := now.Add(-15 * time.Second)
+65 -16
View File
@@ -559,10 +559,26 @@ type ClusterInfo struct {
// PBSDiagnostic contains diagnostic info for a PBS instance // PBSDiagnostic contains diagnostic info for a PBS instance
type PBSDiagnostic struct { type PBSDiagnostic struct {
ID string `json:"id"` ID string `json:"id"`
Name string `json:"name"` Name string `json:"name"`
Host string `json:"host"` Host string `json:"host"`
AuthMethod string `json:"authMethod"` 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"` Connected bool `json:"connected"`
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
ErrorKind string `json:"errorKind,omitempty"` ErrorKind string `json:"errorKind,omitempty"`
@@ -943,10 +959,32 @@ func applyNodeDiagnosticFailure(diag *NodeDiagnostic, failure diagnosticsConnect
} }
func applyPBSDiagnosticFailure(diag *PBSDiagnostic, failure diagnosticsConnectionFailure) { func applyPBSDiagnosticFailure(diag *PBSDiagnostic, failure diagnosticsConnectionFailure) {
diag.Connected = false diag.Probe.Connected = false
diag.Error = failure.message diag.Probe.Error = failure.message
diag.ErrorKind = failure.kind diag.Probe.ErrorKind = failure.kind
diag.Troubleshooting = failure.troubleshooting 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 { 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) 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 { for _, pbsNode := range r.config.PBSInstances {
pbsDiag := PBSDiagnostic{ pbsDiag := PBSDiagnostic{
ID: pbsNode.Name, ID: pbsNode.Name,
@@ -1066,11 +1118,9 @@ func (r *Router) computeDiagnostics(ctx context.Context) DiagnosticsInfo {
Host: pbsNode.Host, Host: pbsNode.Host,
AuthMethod: diagnosticsAuthMethod(pbsNode.User, pbsNode.Password, pbsNode.TokenName, pbsNode.TokenValue), AuthMethod: diagnosticsAuthMethod(pbsNode.User, pbsNode.Password, pbsNode.TokenName, pbsNode.TokenValue),
} }
pbsKeyName := pbsNode.Name if connection, ok := pbsConnections[pbsNode.Name]; ok {
if strings.TrimSpace(pbsKeyName) == "" { applyPBSConnectionDiagnostic(&pbsDiag, connection)
pbsKeyName = pbsNode.Host
} }
monitorConnected, hasMonitorStatus := diagnosticsMonitorConnectionStatus(r.monitor, "pbs-"+pbsKeyName)
if pbsDiag.AuthMethod == "none" { if pbsDiag.AuthMethod == "none" {
applyPBSDiagnosticFailure(&pbsDiag, missingStoredSecretDiagnosticsFailure()) applyPBSDiagnosticFailure(&pbsDiag, missingStoredSecretDiagnosticsFailure())
diag.PBS = append(diag.PBS, pbsDiag) 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") failure := classifyDiagnosticsConnectionFailure(err, "Failed to initialize connection")
applyPBSDiagnosticFailure(&pbsDiag, failure) applyPBSDiagnosticFailure(&pbsDiag, failure)
log.Error().Err(err).Str("pbs", pbsNode.Name).Msg("Diagnostics: PBS client init failed") 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 { } else {
if version, err := client.GetVersion(ctx); err != nil { if version, err := client.GetVersion(ctx); err != nil {
failure := classifyDiagnosticsConnectionFailure(err, "Connection established but version check failed") failure := classifyDiagnosticsConnectionFailure(err, "Connection established but version check failed")
applyPBSDiagnosticFailure(&pbsDiag, failure) applyPBSDiagnosticFailure(&pbsDiag, failure)
log.Error().Err(err).Str("pbs", pbsNode.Name).Msg("Diagnostics: PBS version check failed") 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 { } else {
pbsDiag.Connected = true pbsDiag.Probe.Connected = true
pbsDiag.Details = &PBSDetails{Version: version.Version} pbsDiag.Details = &PBSDetails{Version: version.Version}
pbsDiag.Probe.Details = pbsDiag.Details
} }
} }
+4 -4
View File
@@ -134,13 +134,13 @@ func TestComputeDiagnosticsClassifiesMissingNodeSecrets(t *testing.T) {
if got := diag.PBS[0].AuthMethod; got != "none" { if got := diag.PBS[0].AuthMethod; got != "none" {
t.Fatalf("PBS auth method = %q, want none", got) 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) t.Fatalf("PBS error kind = %q, want no_stored_secret", got)
} }
if !strings.Contains(diag.PBS[0].Error, "no stored credentials") { if !strings.Contains(diag.PBS[0].Probe.Error, "no stored credentials") {
t.Fatalf("PBS error should describe missing credentials, got %q", diag.PBS[0].Error) 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") t.Fatal("expected PBS troubleshooting guidance")
} }
} }
+52
View File
@@ -7,6 +7,8 @@ import (
"testing" "testing"
"time" "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" "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) { func TestClassifyDiagnosticsConnectionFailure(t *testing.T) {
cases := []struct { cases := []struct {
name string name string
@@ -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) { func TestMultiTenantMonitorListOrganizationIDsFallsBackToDefault(t *testing.T) {
got, err := (&MultiTenantMonitor{}).ListOrganizationIDs() got, err := (&MultiTenantMonitor{}).ListOrganizationIDs()
if err != nil { if err != nil {
+16 -5
View File
@@ -4,6 +4,7 @@ import (
"time" "time"
"github.com/rcourtman/pulse-go-rewrite/internal/config" "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/internal/monitoring/errors"
"github.com/rcourtman/pulse-go-rewrite/pkg/pbs" "github.com/rcourtman/pulse-go-rewrite/pkg/pbs"
"github.com/rcourtman/pulse-go-rewrite/pkg/pmg" "github.com/rcourtman/pulse-go-rewrite/pkg/pmg"
@@ -106,14 +107,24 @@ func (m *Monitor) initPBSClients(cfg *config.Config) {
Bool("hasPassword", pbsInst.Password != ""). Bool("hasPassword", pbsInst.Password != "").
Bool("hasToken", pbsInst.TokenValue != ""). Bool("hasToken", pbsInst.TokenValue != "").
Msg("Failed to create PBS client - node will show as disconnected") Msg("Failed to create PBS client - node will show as disconnected")
// Set initial connection health to false for this node m.recordTaskResult(InstanceTypePBS, pbsInst.Name, monErr)
m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, false) 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 continue
} }
m.pbsClients[pbsInst.Name] = client m.pbsClients[pbsInst.Name] = client
log.Info().Str("instance", pbsInst.Name).Msg("PBS client created successfully") log.Info().
// Set initial connection health to true Str("instance", pbsInst.Name).
m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, true) Msg("PBS client created; awaiting first connectivity poll")
} }
} }
@@ -5,6 +5,8 @@ import (
"time" "time"
"github.com/rcourtman/pulse-go-rewrite/internal/config" "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/pbs"
"github.com/rcourtman/pulse-go-rewrite/pkg/proxmox" "github.com/rcourtman/pulse-go-rewrite/pkg/proxmox"
"github.com/rs/zerolog/log" "github.com/rs/zerolog/log"
@@ -154,21 +156,33 @@ func (m *Monitor) retryFailedConnections(ctx context.Context) {
clientConfig.Timeout = 60 * time.Second clientConfig.Timeout = 60 * time.Second
client, err := pbs.NewClient(clientConfig) client, err := pbs.NewClient(clientConfig)
if err != nil { if err != nil {
monErr := monitorerrors.WrapConnectionError("recreate_pbs_client", pbsInst.Name, err)
log.Warn(). log.Warn().
Err(err). Err(monErr).
Str("instance", pbsInst.Name). Str("instance", pbsInst.Name).
Msg("Failed to reconnect PBS client, will retry") 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 continue
} }
m.mu.Lock() m.mu.Lock()
m.pbsClients[pbsInst.Name] = client m.pbsClients[pbsInst.Name] = client
m.mu.Unlock() m.mu.Unlock()
m.setProviderConnectionHealth(InstanceTypePBS, pbsInst.Name, true)
log.Info(). log.Info().
Str("instance", pbsInst.Name). Str("instance", pbsInst.Name).
Msg("Successfully reconnected PBS client") Msg("PBS client recreated; awaiting connectivity poll")
} }
} }
} }
@@ -139,6 +139,7 @@ func TestPollPBSInstance(t *testing.T) {
}, },
state: models.NewState(), state: models.NewState(),
stalenessTracker: NewStalenessTracker(nil), // Pass nil or mock PollMetrics stalenessTracker: NewStalenessTracker(nil), // Pass nil or mock PollMetrics
pollStatusMap: make(map[string]*pollStatus),
nodePendingUpdatesCache: make(map[string]pendingUpdatesCache), nodePendingUpdatesCache: make(map[string]pendingUpdatesCache),
} }
@@ -63,6 +63,12 @@ func TestMonitor_PollPBSInstance_AuthFailure(t *testing.T) {
if snapshot.ConnectionHealth["pbs-pbs-auth-fail"] { if snapshot.ConnectionHealth["pbs-pbs-auth-fail"] {
t.Error("Expected connection health to be false") 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 // Regression: pollStatusMap must record the failure. A defer-arg bug
// previously captured pollErr at register-time (always nil), so failed // previously captured pollErr at register-time (always nil), so failed
@@ -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")
}
+80 -50
View File
@@ -167,15 +167,37 @@ func pbsJobHealthPosture(fact pbs.JobHealthEvidence, freshness models.PBSJobHeal
// pollPBSInstance polls a single PBS instance // pollPBSInstance polls a single PBS instance
func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, client *pbs.Client) { func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, client *pbs.Client) {
defer recoverFromPanic(fmt.Sprintf("pollPBSInstance-%s", instanceName))
start := time.Now() start := time.Now()
debugEnabled := logging.IsLevelEnabled(zerolog.DebugLevel) debugEnabled := logging.IsLevelEnabled(zerolog.DebugLevel)
var pollErr error var pollErr error
var pbsInst models.PBSInstance
publishResult := false
if m.pollMetrics != nil { if m.pollMetrics != nil {
m.pollMetrics.IncInFlight("pbs") 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{ m.pollMetrics.RecordResult(PollResult{
InstanceName: instanceName, InstanceName: instanceName,
InstanceType: "pbs", InstanceType: "pbs",
@@ -184,35 +206,8 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
StartTime: start, StartTime: start,
EndTime: time.Now(), 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 // Get instance config
var instanceCfg *config.PBSInstance var instanceCfg *config.PBSInstance
@@ -229,6 +224,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
} }
} }
if instanceCfg == nil { if instanceCfg == nil {
pollErr = fmt.Errorf("PBS instance config %q not found", instanceName)
log.Error().Str("instance", instanceName).Msg("PBS instance config not found") log.Error().Str("instance", instanceName).Msg("PBS instance config not found")
return return
} }
@@ -240,7 +236,7 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
} }
// Initialize PBS instance with default values // Initialize PBS instance with default values
pbsInst := models.PBSInstance{ pbsInst = models.PBSInstance{
ID: "pbs-" + instanceName, ID: "pbs-" + instanceName,
Name: instanceName, Name: instanceName,
Host: instanceCfg.Host, Host: instanceCfg.Host,
@@ -250,6 +246,23 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
ConnectionHealth: "unhealthy", ConnectionHealth: "unhealthy",
LastSeen: time.Now(), 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 // Try to get version first
version, versionErr := client.GetVersion(ctx) version, versionErr := client.GetVersion(ctx)
@@ -258,7 +271,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
pbsInst.Version = version.Version pbsInst.Version = version.Version
pbsInst.ConnectionHealth = "healthy" pbsInst.ConnectionHealth = "healthy"
m.resetAuthFailures(instanceName, "pbs") m.resetAuthFailures(instanceName, "pbs")
m.setProviderConnectionHealth(InstanceTypePBS, instanceName, true)
if debugEnabled { if debugEnabled {
log.Debug(). log.Debug().
@@ -281,7 +293,6 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
pbsInst.Version = "connected" pbsInst.Version = "connected"
pbsInst.ConnectionHealth = "healthy" pbsInst.ConnectionHealth = "healthy"
m.resetAuthFailures(instanceName, "pbs") m.resetAuthFailures(instanceName, "pbs")
m.setProviderConnectionHealth(InstanceTypePBS, instanceName, true)
log.Info(). log.Info().
Str("instance", instanceName). Str("instance", instanceName).
@@ -289,15 +300,18 @@ func (m *Monitor) pollPBSInstance(ctx context.Context, instanceName string, clie
} else { } else {
pbsInst.Status = "offline" pbsInst.Status = "offline"
pbsInst.ConnectionHealth = "error" 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 pollErr = monErr
log.Error().Err(monErr).Str("instance", instanceName).Msg("failed to connect to PBS") 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) { if errors.IsAuthError(versionErr) || errors.IsAuthError(datastoreErr) {
m.recordAuthFailure(instanceName, "pbs") 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()) 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 // Convert PBS datastores to Storage entries for unified storage view
if len(pbsInst.Datastores) > 0 && instanceCfg.MonitorDatastores { if len(pbsInst.Datastores) > 0 && instanceCfg.MonitorDatastores {
var pbsStorages []models.Storage 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") Msg("Added PBS datastores to unified storage view")
} }
if m.alertManager != nil {
m.alertManager.CheckPBS(pbsInst)
}
// Poll backups if enabled // Poll backups if enabled
if instanceCfg.MonitorBackups { if instanceCfg.MonitorBackups {
if len(pbsInst.Datastores) == 0 { 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 // pollPMGInstance polls a single Proxmox Mail Gateway instance
func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) { func (m *Monitor) pollPMGInstance(ctx context.Context, instanceName string, client *pmg.Client) {
defer recoverFromPanic(fmt.Sprintf("pollPMGInstance-%s", instanceName)) defer recoverFromPanic(fmt.Sprintf("pollPMGInstance-%s", instanceName))