mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-14 16:20:03 +00:00
Fix PBS connection health authority
This commit is contained in:
+92
@@ -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.
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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 {
|
||||||
|
|||||||
@@ -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")
|
||||||
|
}
|
||||||
@@ -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))
|
||||||
|
|||||||
Reference in New Issue
Block a user