From 712f1a3bfeff8f78de242b97fc66526d8f7c5761 Mon Sep 17 00:00:00 2001 From: Pulse Test Date: Sun, 30 Aug 2026 15:35:10 +0100 Subject: [PATCH] Add availability history and fleet view --- docs/AVAILABILITY_HISTORY_CONTRACT.md | 24 +- .../v6/internal/SERVICE_ASSURANCE_PLAN.md | 10 +- docs/release-control/v6/internal/status.json | 2 +- .../v6/internal/subsystems/agent-lifecycle.md | 11 + .../v6/internal/subsystems/api-contracts.md | 11 + .../subsystems/frontend-primitives.md | 12 + .../v6/internal/subsystems/monitoring.md | 15 + .../subsystems/performance-and-scalability.md | 8 + .../internal/subsystems/storage-recovery.md | 8 + .../internal/subsystems/unified-resources.md | 9 + frontend-modern/browser-verification.json | 46 +- .../api/__tests__/availabilityHistory.test.ts | 49 + .../src/api/availabilityHistory.ts | 90 ++ .../src/api/availabilityTargets.ts | 1 + .../standalone/AvailabilityChecksTable.tsx | 444 +++++---- .../standalone/AvailabilityFleetView.tsx | 325 +++++++ .../standalone/StandalonePageSurface.tsx | 18 +- .../__tests__/AvailabilityFleetView.test.tsx | 147 +++ .../routing/__tests__/resourceLinks.test.ts | 2 +- frontend-modern/src/routing/resourceLinks.ts | 1 + internal/api/availability_handlers.go | 13 +- internal/api/availability_handlers_test.go | 24 + internal/api/availability_history_handlers.go | 257 ++++++ .../api/availability_history_handlers_test.go | 95 ++ internal/api/route_inventory_test.go | 1 + internal/api/router_routes_monitoring.go | 1 + internal/api/security_regression_test.go | 1 + internal/config/availability.go | 23 + internal/config/availability_test.go | 34 + internal/hostagent/availability.go | 13 +- internal/monitoring/availability_poller.go | 54 +- .../monitoring/availability_probe_agent.go | 54 +- .../availability_probe_agent_test.go | 52 ++ pkg/agents/host/report.go | 14 +- pkg/agents/host/report_test.go | 6 +- pkg/metrics/availability_history.go | 843 ++++++++++++++++++ pkg/metrics/availability_history_test.go | 212 +++++ pkg/metrics/store.go | 31 +- pkg/metrics/store_query_plan_test.go | 124 +++ 39 files changed, 2843 insertions(+), 242 deletions(-) create mode 100644 frontend-modern/src/api/__tests__/availabilityHistory.test.ts create mode 100644 frontend-modern/src/api/availabilityHistory.ts create mode 100644 frontend-modern/src/features/standalone/AvailabilityFleetView.tsx create mode 100644 frontend-modern/src/features/standalone/__tests__/AvailabilityFleetView.test.tsx create mode 100644 internal/api/availability_history_handlers.go create mode 100644 internal/api/availability_history_handlers_test.go create mode 100644 pkg/metrics/availability_history.go create mode 100644 pkg/metrics/availability_history_test.go diff --git a/docs/AVAILABILITY_HISTORY_CONTRACT.md b/docs/AVAILABILITY_HISTORY_CONTRACT.md index b38e2df88..e7b2fa6a4 100644 --- a/docs/AVAILABILITY_HISTORY_CONTRACT.md +++ b/docs/AVAILABILITY_HISTORY_CONTRACT.md @@ -1,6 +1,6 @@ # Availability History and Fleet View Contract -Status: Build-ready +Status: Implemented (Slice A) Date: 2026-08-30 Scope: `pulse` only Demand owner: `pulse-pro/FEATURE_REQUESTS.md`, "Fleet-scale machine @@ -251,3 +251,25 @@ The fleet slice additionally requires a rendered-browser contract at 50 targets, keyboard access to every tile, non-color state labels, and proof that an API failure renders "history unavailable" without changing current target health. + +## Implementation record + +Slice A was implemented on 2026-08-30 through the source-owned +`availability_observations`, `availability_history_buckets`, and +`availability_revision_boundaries` tables in the shared metrics-store +lifecycle; the monitoring-owned local and assigned-agent ingest paths; the +bounded `POST /api/availability-history` read contract; and the URL-owned +table/fleet presentation under **Machines -> Availability**. + +The executable proof surface is: + +- `pkg/metrics/availability_history_test.go` for categorical duration, + coverage, privacy, idempotency, restart, revision, retention, deletion, and + 200-target bounds; +- `internal/monitoring/availability_probe_agent_test.go` for assigned-agent + receipt-time authority, duplicate retry, and stale-revision rejection; +- `internal/api/availability_history_handlers_test.go` and the API route/scope + contracts for bounded reads and authorization; +- `frontend-modern/src/features/standalone/__tests__/AvailabilityFleetView.test.tsx` + for a rendered 50-target fleet, keyboard-operable detail entry, non-color + labels, and the history-failure boundary. diff --git a/docs/release-control/v6/internal/SERVICE_ASSURANCE_PLAN.md b/docs/release-control/v6/internal/SERVICE_ASSURANCE_PLAN.md index 4011b6bf7..8c380992e 100644 --- a/docs/release-control/v6/internal/SERVICE_ASSURANCE_PLAN.md +++ b/docs/release-control/v6/internal/SERVICE_ASSURANCE_PLAN.md @@ -1,7 +1,7 @@ # Infrastructure-Aware Service Assurance Plan Last updated: 2026-08-30 -Status: PROPOSED +Status: ACCEPTED — SLICE A IMPLEMENTED Governance surfaces: - `status.json.coverage_gaps.infrastructure-aware-service-assurance` @@ -224,6 +224,14 @@ Exit conditions: - the UI never presents missing observations as healthy time or labels the result as an SLA. +Implementation record (2026-08-30): Slice A is delivered through the +monitoring-owned categorical history store and rollups, server-authored +configuration revisions and remote receipt timeline, the bounded +`/api/availability-history` batch contract, and the URL-owned Availability +fleet presentation. Its release proofs live in the owner contract. Slices B +through H remain ordered future work; acceptance of this product lane does not +imply that deferred breadth is already delivered. + ### Slice B: Application Response Contracts Deliver a bounded HTTP and HTTPS verification contract attached to a known diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index dc2db7a54..d3673f31f 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -9891,7 +9891,7 @@ "id": "infrastructure-aware-service-assurance", "name": "Infrastructure-Aware Service Assurance", "summary": "Extend Pulse from infrastructure-state monitoring into service-delivery assurance by attaching active application verification to canonical resources, preserving coverage-aware history, using discovered services and remote agents for setup and observation, correlating failures through infrastructure relationships, and presenting the resulting evidence through fleet, incident, timeline, and Patrol workflows instead of a standalone monitor clone.", - "status": "proposed", + "status": "accepted", "recorded_at": "2026-08-30", "target_id": "v6-product-lane-expansion", "current_lane_ids": [ diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index dff4265b5..d1ad006c6 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -6166,6 +6166,17 @@ or discarding observations that the primary never received. The scheduler and concurrent-enqueue coverage in `internal/hostagent/availability_test.go` pins the ordering and acknowledgement contract. +Every assigned availability result also carries the server-issued target +configuration revision and a stable observation ID. The agent allocates that +ID once when the scheduled execution completes and preserves it across report +buffering or retry, so an acknowledged report retry cannot create a second +history observation. The server rejects results for an obsolete revision, +uses receipt time as the coverage timeline, and retains the agent-authored +check time only as evidence metadata. Legacy agents that omit the additive ID +remain ingestible through a deterministic server fallback, but they do not +gain authority to author revision boundaries, target identity, retention, or +the service-assurance timeline. + Observer configuration is explicit, versioned, and file-backed. It contains no raw token values and resolves each token from a separate private absolute-path file. Proxmox registration is also destination-scoped: the primary retains its diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index 3d84a93d7..088c5e013 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -8758,6 +8758,17 @@ Mock availability fixtures must still behave like saved targets: `/api/connectio reports them as availability rows, `/api/availability-targets` lists them with probe status, and saved-test calls return the synthetic probe result instead of attempting live network I/O against demo-only addresses. +Availability history is an additive bounded read contract at +`POST /api/availability-history`, protected by `monitoring:read`. Callers send +one allowlisted range and at most 200 unique saved-target IDs; the server +returns at most 120 presentation buckets per target plus categorical duration +coverage, reachable-only latency, and server-authored configuration-revision +boundaries. The store performs one batch read rather than a query per target, +and per-target absence remains an explicit `not_found` result instead of +fabricated unknown history. Responses never include target addresses, raw +errors, agent identity, certificate detail, or customer identity. Mock mode +preserves the same response shape so the browser exercises the real transport +contract without making network probes. Unified-resource transport adds typed availability trust fields without changing the saved-target CRUD owner. `availability` remains the singular compatibility summary; `availabilityChecks` is the complete attached set; each diff --git a/docs/release-control/v6/internal/subsystems/frontend-primitives.md b/docs/release-control/v6/internal/subsystems/frontend-primitives.md index b72070d8a..f697de71f 100644 --- a/docs/release-control/v6/internal/subsystems/frontend-primitives.md +++ b/docs/release-control/v6/internal/subsystems/frontend-primitives.md @@ -6375,6 +6375,18 @@ frontend-primitives-owned Machines surface as a focused Availability checks tab rather than a new primary nav item. The page may show availability checks beside standalone Pulse Agent machines, but Settings remains the add/edit owner and the app shell must not add a separate top-level Availability destination. +That Availability checks tab owns a URL-addressable `view=fleet` presentation +alongside the existing table; table remains the default and `q` plus `status` +filters are shared between both modes. Fleet tiles combine the canonical +current-health projection with the bounded history batch: categorical state +coverage is labelled in text as reachable, unreachable, indeterminate, or +unknown, and latency is drawn only for reachable evidence so gaps cannot be +misread as zero latency. A tile opens the existing `ResourceDetailDrawer` +rather than a service-monitor-specific detail model. History failure must stay +inside the fleet surface with explicit copy while current status and resource +navigation remain usable. Desktop and phone layouts must keep every tile +keyboard-operable, preserve the textual legend, and avoid horizontal clipping +at fleet scale. The Machines page must not pretend its machine list is a generic overview: the default tab is `Machines`, the Machines table is only for Pulse Agent-backed resources with host telemetry, and the full availability-check row list belongs diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index fcd18410f..f97a78705 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -1443,6 +1443,21 @@ load from the config persistence boundary, schedule through the default low-overhead check, while TCP and HTTP are canonical fallbacks for devices or runtimes where ICMP is unavailable or the useful signal is a port or web interface. +Completed scheduled availability probes also write source-owned categorical +history through the metrics-store writer lifecycle without encoding outcomes +as ordinary numeric metrics. The durable vocabulary is `reachable`, +`unreachable`, and `indeterminate`; uncovered time is derived as `unknown` +from the bounded validity window rather than backfilled from current status, +alerts, or stale-state synthesis. Local checks use the server-authored check +time. Accepted assigned-agent results use server receipt time for coverage, +retain the agent time only as evidence metadata, and carry a stable observation +ID so retries are idempotent. Execution-defining target edits increment a +server-authored configuration revision and history exposes the revision +boundary. Raw and minute/hour/day rollups share the existing entitlement and +retention lifecycle, are deleted with the target, and retain no target address, +agent identity, raw error, certificate detail, or customer identity. The +monitoring-read batch path is bounded to 200 targets and 120 presentation +buckets without a query per target. HTTPS checks also author one canonical certificate observation from the same probe execution. `internal/availabilityprobe` captures the presented leaf and `pkg/tlsutil/certificate.go` derives subject, issuer, SANs, SHA-256 fingerprint, diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index aaefbf2c2..3dc2bb01f 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -2642,6 +2642,14 @@ store-backed disk history path as longer-term disk charts, with the backend doing range selection and fallback. Feature-local polling loops or browser-side disk ring buffers are forbidden because they duplicate live sampling work and drift out of sync with the governed history timeline. +Categorical availability history shares that metrics-store lifecycle without +being encoded as numeric metric samples. Raw observations enter through the +single writer queue, compact into exact-duration minute/hour/day buckets, and +prune on the existing entitlement-aware retention worker. A fleet read accepts +at most 200 target IDs and 120 output buckets and completes with three bounded +queries for rollups, live tails, and revision boundaries, never one query per +tile. Target/timeline, tier/retention, and revision/time indexes are part of +the hot-path contract and are pinned by the metrics-store query-plan proof. That same hot-path ownership now also requires lazy-load-safe websocket consumption. `frontend-modern/src/components/Workloads/useWorkloadsState.ts` may read connection and alert state only through diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index 37d061366..5b2987108 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -5622,6 +5622,14 @@ automated remediation through the shared action lifecycle, while recovery evidence remains available for operator review and for restoration to active monitoring. +Availability history stored beside metrics is monitoring evidence, not backup +inventory, a recovery point, restore evidence, or storage-action authority. +Its raw and minute/hour/day tiers follow the metrics store's +entitlement-aware retention lifecycle, and deleting a saved availability +target removes its observations, rollups, and revision boundaries as one owned +target lifecycle operation. Those rules must not delete unified-resource +change history, provider snapshots, backups, or any recovery evidence. + ### Configuration archive recovery is authorized before persistence Encrypted configuration export/import is a storage-recovery boundary only diff --git a/docs/release-control/v6/internal/subsystems/unified-resources.md b/docs/release-control/v6/internal/subsystems/unified-resources.md index 5bbbf6f6e..9105b06c3 100644 --- a/docs/release-control/v6/internal/subsystems/unified-resources.md +++ b/docs/release-control/v6/internal/subsystems/unified-resources.md @@ -4644,6 +4644,15 @@ extend those owners rather than recreating `frontend-modern/src/features/infrastructure/`, a `/infrastructure` route, or a separate top-level availability route. +The Availability checks table and fleet view are two presentations of the same +source-owned `network-endpoint` resources. `view=fleet`, `q`, and `status` are +URL presentation state only: they must not mint another resource kind, clone +saved targets, or infer relationships from history buckets. Fleet tiles open +the shared resource drawer by canonical resource ID, while +`/api/availability-history` keys evidence by the saved availability target ID +already carried by that resource. History absence or failure cannot overwrite +the unified resource's current status, correlation, source ownership, or +navigation identity. Shared unified-resource consumers now also normalize org scope through `frontend-modern/src/utils/orgScope.ts` before building cache keys or multi-tenant resource fetch state, so the canonical resource hooks do not diff --git a/frontend-modern/browser-verification.json b/frontend-modern/browser-verification.json index d8ffa13d9..d0cf6ecc6 100644 --- a/frontend-modern/browser-verification.json +++ b/frontend-modern/browser-verification.json @@ -1,23 +1,27 @@ { "version": 1, - "base_sha": "7e92ac8118a7661c52469cb29a5a0b9ffb2e27d4", - "verified_at": "2026-08-30T13:30:20Z", + "base_sha": "a7f1a5e25ffd636c98ee51065c2fe0f0dc7086a6", + "verified_at": "2026-08-30T14:28:38Z", "result": "passed", "changed_paths": [ - "frontend-modern/src/components/Workloads/WorkloadsSurface.tsx", - "frontend-modern/src/components/Workloads/workloadsFilterModel.ts", - "frontend-modern/src/features/proxmox/ProxmoxPageSurface.tsx", - "frontend-modern/src/features/vmware/VmwarePageSurface.tsx" + "frontend-modern/src/api/availabilityHistory.ts", + "frontend-modern/src/api/availabilityTargets.ts", + "frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx", + "frontend-modern/src/features/standalone/AvailabilityFleetView.tsx", + "frontend-modern/src/features/standalone/StandalonePageSurface.tsx", + "frontend-modern/src/routing/resourceLinks.ts" ], "content_sha256": { - "frontend-modern/src/components/Workloads/WorkloadsSurface.tsx": "dfd5e844f1261193811d05529a6cf796f20b0d403feffaa322bc16a0126ec247", - "frontend-modern/src/components/Workloads/workloadsFilterModel.ts": "c177b60a48e679e8bede6d14b20cbdbd8a13e743a0e8c058c0060e232091a977", - "frontend-modern/src/features/proxmox/ProxmoxPageSurface.tsx": "516b1a43169d7f3191f3d101dbf9140fe36c1c0099db45ef76222dac6e0210c9", - "frontend-modern/src/features/vmware/VmwarePageSurface.tsx": "524832fce536f4d708246f9b2a7d2ea35e1fb059aa624bc05ce53a54366f6f67" + "frontend-modern/src/api/availabilityHistory.ts": "1fea7ef0612b872c387213e590b47fe42ecdb87ca2678b7b01b546273da65b0d", + "frontend-modern/src/api/availabilityTargets.ts": "549826b1af427820b9d29d53cc68d895ae19a3e1a9426485fc34903972c55109", + "frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx": "eef59dc4eea1e7c4029cf3e89c29c7a7ed26484a67bdea213bfdc019ed3bb10e", + "frontend-modern/src/features/standalone/AvailabilityFleetView.tsx": "e6a21fcd193c51b3e0d164ca549821c6d2bf34c92c45fe884b019b56f3c87127", + "frontend-modern/src/features/standalone/StandalonePageSurface.tsx": "20d3a5c2107c5510d579e2456bd1aba741137f05acff8acca6805dc8fd387563", + "frontend-modern/src/routing/resourceLinks.ts": "dee9a426de785e23390ba49c9067f55c8f923cecfce1247ff18f9b1004e0cc90" }, "routes": [ - "/proxmox/overview", - "/vmware/overview" + "/standalone/availability?view=fleet", + "/standalone/availability" ], "viewports": [ { @@ -30,15 +34,17 @@ } ], "states": [ - "Existing Proxmox Bars and History states retained the previously verified shared workload presentation after moving the identical accessors behind the canonical binding", - "vSphere workload composition supplied display, hover, range, and discovery state through the same atomic binding in the rendered platform contract", - "The generic WorkloadsSurface supplied the complete binding without a page-local partial metric prop list", - "The shared filter retained its desktop and mobile Details and History preference behavior" + "Fleet view rendered all seven mock availability targets with current status, method, source, last check, categorical state history, reachable-only latency history, coverage text, and revision boundaries", + "The textual Reachable, Unreachable, Indeterminate, and Unknown legend remained visible without relying on color alone", + "The resource detail drawer opened from an availability tile and preserved the existing canonical resource detail presentation", + "The solar search state reduced the fleet to one matching target and persisted q=solar in the URL", + "Table view restored all seven availability rows with the fleet parameter removed, and switching back restored view=fleet", + "The 390 by 844 layout kept the fleet controls usable and stacked availability tiles into a readable single column" ], "interactions": [ - "rechecked the shared workload history interaction already exercised on Proxmox against the unchanged WorkloadsFilter and guest-row implementation", - "rendered the vSphere platform contract and verified its workload toolbar now receives the complete canonical binding", - "ran the shared WorkloadsFilter interaction suite for Bars, Trends, Details, History, range selection, and responsive controls", - "ran structural contracts that reject partial metric-control wiring in Proxmox, vSphere, and the generic WorkloadsSurface" + "opened an availability tile and dismissed the canonical resource detail drawer", + "entered and cleared the solar search filter while checking the URL-owned query state", + "switched from Fleet to Table and back through the shared View control while checking URL and row or tile counts", + "repeated the fleet inspection at desktop and narrow viewports and checked scrolling, clipping, history labels, and control placement" ] } diff --git a/frontend-modern/src/api/__tests__/availabilityHistory.test.ts b/frontend-modern/src/api/__tests__/availabilityHistory.test.ts new file mode 100644 index 000000000..a96ff1c7a --- /dev/null +++ b/frontend-modern/src/api/__tests__/availabilityHistory.test.ts @@ -0,0 +1,49 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { + AvailabilityHistoryAPI, + type AvailabilityHistoryResponse, +} from '@/api/availabilityHistory'; +import { apiFetchJSON } from '@/utils/apiClient'; + +vi.mock('@/utils/apiClient', () => ({ apiFetchJSON: vi.fn() })); + +const mockedApiFetchJSON = vi.mocked(apiFetchJSON); + +beforeEach(() => { + mockedApiFetchJSON.mockReset(); +}); + +describe('AvailabilityHistoryAPI', () => { + it('deduplicates target ids and posts one bounded batch', async () => { + const response: AvailabilityHistoryResponse = { + start: '2026-08-29T12:00:00Z', + end: '2026-08-30T12:00:00Z', + targets: [], + }; + mockedApiFetchJSON.mockResolvedValue(response); + + await AvailabilityHistoryAPI.batch(['one', ' one ', '', 'two']); + + expect(mockedApiFetchJSON).toHaveBeenCalledWith('/api/availability-history', { + method: 'POST', + body: JSON.stringify({ targetIds: ['one', 'two'], range: '24h' }), + }); + }); + + it('chunks fleets larger than the server bound without target-by-target reads', async () => { + mockedApiFetchJSON.mockImplementation(async (_path, init) => { + const request = JSON.parse(String(init?.body)) as { targetIds: string[] }; + return { + start: '2026-08-29T12:00:00Z', + end: '2026-08-30T12:00:00Z', + targets: request.targetIds.map((targetId) => ({ targetId })), + } satisfies AvailabilityHistoryResponse; + }); + const ids = Array.from({ length: 450 }, (_, index) => `target-${index}`); + + const response = await AvailabilityHistoryAPI.batch(ids); + + expect(mockedApiFetchJSON).toHaveBeenCalledTimes(3); + expect(response.targets).toHaveLength(450); + }); +}); diff --git a/frontend-modern/src/api/availabilityHistory.ts b/frontend-modern/src/api/availabilityHistory.ts new file mode 100644 index 000000000..3afb8297d --- /dev/null +++ b/frontend-modern/src/api/availabilityHistory.ts @@ -0,0 +1,90 @@ +import { apiFetchJSON } from '@/utils/apiClient'; + +const AVAILABILITY_HISTORY_PATH = '/api/availability-history'; +const MAX_TARGETS_PER_REQUEST = 200; + +export interface AvailabilityLatencySummary { + average: number; + min: number; + max: number; +} + +export interface AvailabilityHistorySummary { + reachableSeconds: number; + unreachableSeconds: number; + indeterminateSeconds: number; + unknownSeconds: number; + coveragePercent: number; + availabilityPercent?: number; + reachableLatencyMillis?: AvailabilityLatencySummary; +} + +export interface AvailabilityHistoryBucket { + start: string; + end: string; + reachableSeconds: number; + unreachableSeconds: number; + indeterminateSeconds: number; + unknownSeconds: number; + latencyMillis?: AvailabilityLatencySummary; +} + +export interface AvailabilityRevisionBoundary { + revision: number; + at: string; +} + +export interface AvailabilityHistoryTargetError { + code: 'not_found' | 'forbidden' | string; + message: string; +} + +export interface AvailabilityHistoryTarget { + targetId: string; + summary?: AvailabilityHistorySummary; + buckets?: AvailabilityHistoryBucket[]; + revisionBoundaries?: AvailabilityRevisionBoundary[]; + error?: AvailabilityHistoryTargetError; +} + +export interface AvailabilityHistoryResponse { + start: string; + end: string; + targets: AvailabilityHistoryTarget[]; +} + +const uniqueTargetIds = (targetIds: readonly string[]): string[] => [ + ...new Set(targetIds.map((targetId) => targetId.trim()).filter(Boolean)), +]; + +export class AvailabilityHistoryAPI { + static async batch( + targetIds: readonly string[], + range = '24h', + ): Promise { + const ids = uniqueTargetIds(targetIds); + if (ids.length === 0) { + const now = new Date().toISOString(); + return { start: now, end: now, targets: [] }; + } + + const chunks: string[][] = []; + for (let index = 0; index < ids.length; index += MAX_TARGETS_PER_REQUEST) { + chunks.push(ids.slice(index, index + MAX_TARGETS_PER_REQUEST)); + } + const responses = await Promise.all( + chunks.map((targetIdsChunk) => + apiFetchJSON(AVAILABILITY_HISTORY_PATH, { + method: 'POST', + body: JSON.stringify({ targetIds: targetIdsChunk, range }), + }), + ), + ); + + return { + start: responses[0]?.start ?? new Date().toISOString(), + end: responses[0]?.end ?? new Date().toISOString(), + targets: responses.flatMap((response) => response.targets), + }; + } +} diff --git a/frontend-modern/src/api/availabilityTargets.ts b/frontend-modern/src/api/availabilityTargets.ts index 78e1d7f20..6b6df8d9f 100644 --- a/frontend-modern/src/api/availabilityTargets.ts +++ b/frontend-modern/src/api/availabilityTargets.ts @@ -32,6 +32,7 @@ export interface AvailabilityProbeStatus { export interface AvailabilityTarget { id: string; + configRevision?: number; name: string; targetKind?: AvailabilityTargetKind; address: string; diff --git a/frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx b/frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx index 60e182eab..8a953a337 100644 --- a/frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx +++ b/frontend-modern/src/features/standalone/AvailabilityChecksTable.tsx @@ -1,8 +1,9 @@ import { A } from '@solidjs/router'; -import { Show, createMemo, type Component, type JSX } from 'solid-js'; +import { Show, createMemo, createResource, type Component, type JSX } from 'solid-js'; import PlusIcon from 'lucide-solid/icons/plus'; import SettingsIcon from 'lucide-solid/icons/settings'; import { MetadataBadge } from '@/components/shared/MetadataBadge'; +import { FilterSegmentedControl } from '@/components/shared/FilterToolbar'; import { StatusDot } from '@/components/shared/StatusDot'; import { TableCell, TableHead, TableRow } from '@/components/shared/Table'; import { @@ -29,6 +30,7 @@ import { getPlatformResourceDetailRowClass, } from '@/features/platformPage/PlatformResourceDetailTableRow'; import type { Resource, ResourceAvailabilityMeta } from '@/types/resource'; +import { AvailabilityHistoryAPI } from '@/api/availabilityHistory'; import { getAvailabilityProbeEndpointLabel, getAvailabilityProbePresentation, @@ -42,6 +44,9 @@ import { getStandaloneResourceStatusIndicator, sortStandaloneResourcesByAttention, } from './standalonePageModel'; +import { AvailabilityFleetView } from './AvailabilityFleetView'; + +export type AvailabilityChecksView = 'table' | 'fleet'; const settingsLinkClass = 'inline-flex min-h-8 items-center justify-center gap-1.5 rounded-md border border-border bg-surface px-2.5 py-1 text-xs font-medium text-base-content transition-colors hover:bg-surface-hover'; @@ -73,6 +78,13 @@ export const AvailabilityChecksTable: Component<{ emptyDescription: string; /** Connected agent hosts, used to name the source of probe-reported results. */ probeAgentOptions?: readonly ProbeAgentOption[]; + view?: AvailabilityChecksView; + onViewChange?: (view: AvailabilityChecksView) => void; + externalSearch?: () => string; + onExternalSearchChange?: (value: string) => void; + externalStatus?: () => PlatformResourceStatusFilter; + onExternalStatusChange?: (status: PlatformResourceStatusFilter) => void; + onResetFilters?: () => void; }> = (props) => { const tableState = createPlatformTableFilterState({ resources: () => props.resources, @@ -84,8 +96,45 @@ export const AvailabilityChecksTable: Component<{ if (variant === 'danger') return 'offline'; return 'degraded'; }), + externalSearch: props.externalSearch, + onExternalSearchChange: props.onExternalSearchChange, + externalStatus: props.externalStatus, + onExternalStatusChange: props.onExternalStatusChange, }); + const resetFilters = () => { + if (props.onResetFilters) { + props.onResetFilters(); + return; + } + tableState.resetFilters(); + }; const orderedChecks = createMemo(() => sortStandaloneResourcesByAttention(tableState.filtered())); + const historyTargetIDs = createMemo(() => + props.resources + .map((resource) => availabilityFor(resource)?.targetId) + .filter((targetID): targetID is string => Boolean(targetID)), + ); + const historySource = createMemo(() => + (props.view ?? 'table') === 'fleet' && historyTargetIDs().length > 0 + ? historyTargetIDs().join('\u0000') + : undefined, + ); + const [history, historyActions] = createResource(historySource, async () => { + try { + return { + response: await AvailabilityHistoryAPI.batch(historyTargetIDs(), '24h'), + error: undefined, + }; + } catch (error) { + return { + response: undefined, + error: error instanceof Error ? error.message : 'Availability history is unavailable', + }; + } + }); + const historyByTarget = createMemo( + () => new Map((history()?.response?.targets ?? []).map((target) => [target.targetId, target])), + ); const drawer = createPlatformResourceDetailState({ idPrefix: 'availability-check-detail' }); const resolveResourceLabel = createPlatformResourceLabelResolver(() => props.resources); @@ -121,6 +170,24 @@ export const AvailabilityChecksTable: Component<{ visible={tableState.visible()} total={tableState.total()} rowNoun="checks" + hasActiveFilters={tableState.hasActiveFilters()} + onResetFilters={resetFilters} + viewOptions={ +
+
+ Availability presentation +
+ props.onViewChange?.(value as AvailabilityChecksView)} + options={[ + { value: 'table', label: 'Table' }, + { value: 'fleet', label: 'Fleet' }, + ]} + /> +
+ } /> } > - - - - Add service/device check - - - - Manage - - + void historyActions.refetch()} + /> } - tableClass="min-w-full table-fixed text-xs md:min-w-[900px]" - header={ - <> - - Check - - - Method - - - Target - - - Result - - - - - - - - - } - body={ - - {(check) => { - const availability = () => availabilityFor(check); - const probe = () => getAvailabilityProbePresentation(check); - const indicator = () => getStandaloneResourceStatusIndicator(check); - const method = () => probe()?.methodLabel ?? availability()?.protocol ?? 'Probe'; - const result = () => probe()?.resultLabel ?? indicator().label; - const target = () => formatTarget(check); - const probeSource = () => - getProbeSourceChipLabel( - props.probeAgentOptions ?? [], - availability()?.probeAgentId, - ); - const detailRowId = () => drawer.detailRowId(check); - const isExpanded = () => drawer.isExpanded(check); + > + + + + Add service/device check + + + + Manage + + + } + tableClass="min-w-full table-fixed text-xs md:min-w-[900px]" + header={ + <> + + Check + + + Method + + + Target + + + Result + + + + + + + + + } + body={ + + {(check) => { + const availability = () => availabilityFor(check); + const probe = () => getAvailabilityProbePresentation(check); + const indicator = () => getStandaloneResourceStatusIndicator(check); + const method = () => + probe()?.methodLabel ?? availability()?.protocol ?? 'Probe'; + const result = () => probe()?.resultLabel ?? indicator().label; + const target = () => formatTarget(check); + const probeSource = () => + getProbeSourceChipLabel( + props.probeAgentOptions ?? [], + availability()?.probeAgentId, + ); + const detailRowId = () => drawer.detailRowId(check); + const isExpanded = () => drawer.isExpanded(check); - return ( - <> - drawer.toggle(check)} - onKeyDown={drawer.handleActivationKey(check)} - tabIndex={0} - > - -
- drawer.toggle(check)} - /> - - - {check.name} - -
-
- + drawer.toggle(check)} + onKeyDown={drawer.handleActivationKey(check)} + tabIndex={0} > - {method()} - - - - {target()} - - - - - {result()} - - - {(sourceLabel) => ( - +
+ drawer.toggle(check)} + /> + + - {sourceLabel()} - - )} - - - - - - - - - - drawer.close(check)} - /> - - ); - }} - - } - /> + {check.name} + +
+
+ + {method()} + + + + {target()} + + + + + {result()} + + + {(sourceLabel) => ( + + {sourceLabel()} + + )} + + + + + + + + +
+ drawer.close(check)} + /> + + ); + }} +
+ } + /> +
diff --git a/frontend-modern/src/features/standalone/AvailabilityFleetView.tsx b/frontend-modern/src/features/standalone/AvailabilityFleetView.tsx new file mode 100644 index 000000000..8315ea08f --- /dev/null +++ b/frontend-modern/src/features/standalone/AvailabilityFleetView.tsx @@ -0,0 +1,325 @@ +import { For, Show, createMemo, createSignal, type Component } from 'solid-js'; +import RefreshCwIcon from 'lucide-solid/icons/refresh-cw'; +import { ResourceDetailDrawer } from '@/components/Infrastructure/ResourceDetailDrawer'; +import { Button } from '@/components/shared/Button'; +import { MetadataBadge } from '@/components/shared/MetadataBadge'; +import { StatusDot } from '@/components/shared/StatusDot'; +import type { + AvailabilityHistoryBucket, + AvailabilityHistoryTarget, +} from '@/api/availabilityHistory'; +import type { Resource, ResourceAvailabilityMeta } from '@/types/resource'; +import { formatRelativeTime } from '@/utils/format'; +import { + getAvailabilityProbeEndpointLabel, + getAvailabilityProbePresentation, +} from '@/utils/availabilityProbePresentation'; +import { getProbeSourceChipLabel, type ProbeAgentOption } from '@/utils/availabilityProbeAgents'; +import { getStandaloneResourceStatusIndicator } from './standalonePageModel'; + +type AvailabilityFleetHistoryState = 'reachable' | 'unreachable' | 'indeterminate' | 'unknown'; + +const statePresentation: Record< + AvailabilityFleetHistoryState, + { label: string; className: string } +> = { + reachable: { label: 'Reachable', className: 'bg-emerald-500 dark:bg-emerald-400' }, + unreachable: { label: 'Unreachable', className: 'bg-rose-500 dark:bg-rose-400' }, + indeterminate: { label: 'Indeterminate', className: 'bg-amber-400 dark:bg-amber-300' }, + unknown: { label: 'Unknown', className: 'bg-slate-300 dark:bg-slate-600' }, +}; + +const availabilityFor = (resource: Resource): ResourceAvailabilityMeta | undefined => + resource.availability ?? + (resource.platformData?.availability as ResourceAvailabilityMeta | undefined); + +const historyState = (bucket: AvailabilityHistoryBucket): AvailabilityFleetHistoryState => { + const durations: Array<[AvailabilityFleetHistoryState, number]> = [ + ['unreachable', bucket.unreachableSeconds], + ['indeterminate', bucket.indeterminateSeconds], + ['reachable', bucket.reachableSeconds], + ['unknown', bucket.unknownSeconds], + ]; + durations.sort((left, right) => right[1] - left[1]); + return durations[0]?.[0] ?? 'unknown'; +}; + +const formatPercent = (value: number): string => + `${value.toLocaleString(undefined, { maximumFractionDigits: 2 })}%`; + +const availabilityText = (history: AvailabilityHistoryTarget | undefined): string => { + if (!history?.summary) return 'History unavailable'; + const { summary } = history; + if (summary.coveragePercent < 90 || summary.availabilityPercent === undefined) { + const observed = + summary.reachableSeconds + summary.unreachableSeconds + summary.indeterminateSeconds; + const observedMinutes = Math.round(observed / 60); + return `Insufficient coverage · ${observedMinutes.toLocaleString()}m observed`; + } + return `${formatPercent(summary.availabilityPercent)} available · ${formatPercent(summary.coveragePercent)} observed`; +}; + +const latencyPaths = ( + buckets: readonly AvailabilityHistoryBucket[], +): { path: string; label: string }[] => { + const values = buckets + .map((bucket) => bucket.latencyMillis?.average) + .filter((value): value is number => typeof value === 'number' && Number.isFinite(value)); + if (values.length === 0) return []; + const min = Math.min(...values); + const max = Math.max(...values); + const spread = Math.max(1, max - min); + const denominator = Math.max(1, buckets.length - 1); + const paths: { path: string; label: string }[] = []; + let points: string[] = []; + const flush = () => { + if (points.length > 0) { + paths.push({ path: points.join(' '), label: `${min}–${max} ms reachable latency` }); + points = []; + } + }; + buckets.forEach((bucket, index) => { + const latency = bucket.latencyMillis?.average; + if (typeof latency !== 'number' || !Number.isFinite(latency)) { + flush(); + return; + } + const x = (index / denominator) * 100; + const y = 22 - ((latency - min) / spread) * 18; + points.push(`${x.toFixed(2)},${y.toFixed(2)}`); + }); + flush(); + return paths; +}; + +const AvailabilityStateStrip: Component<{ buckets: readonly AvailabilityHistoryBucket[] }> = ( + props, +) => { + const states = createMemo(() => props.buckets.map(historyState)); + const description = createMemo(() => { + const counts = new Map(); + for (const state of states()) counts.set(state, (counts.get(state) ?? 0) + 1); + return (Object.keys(statePresentation) as AvailabilityFleetHistoryState[]) + .filter((state) => (counts.get(state) ?? 0) > 0) + .map((state) => `${counts.get(state)} ${statePresentation[state].label.toLowerCase()}`) + .join(', '); + }); + return ( + + ); +}; + +const AvailabilityLatencyLine: Component<{ buckets: readonly AvailabilityHistoryBucket[] }> = ( + props, +) => { + const paths = createMemo(() => latencyPaths(props.buckets)); + return ( + 0} + fallback={ +
No reachable latency
+ } + > + + + {(path) => ( + + )} + + +
+ ); +}; + +export const AvailabilityFleetView: Component<{ + resources: readonly Resource[]; + historyByTarget: ReadonlyMap; + historyLoading: boolean; + historyError?: string; + probeAgentOptions?: readonly ProbeAgentOption[]; + onRetryHistory?: () => void; +}> = (props) => { + const [selectedResource, setSelectedResource] = createSignal(); + const resolveResourceLabel = (resourceId: string): string | undefined => + props.resources.find((resource) => resource.id === resourceId)?.name; + + return ( +
+
+
+ + {(presentation) => ( + + + )} + + 24-hour evidence · reachable latency only +
+ +
+ History unavailable. Current health is unchanged. + + + +
+
+
+ +
+ + {(resource) => { + const availability = () => availabilityFor(resource); + const probe = () => getAvailabilityProbePresentation(resource); + const indicator = () => getStandaloneResourceStatusIndicator(resource); + const targetID = () => availability()?.targetId ?? resource.platformId ?? resource.id; + const history = () => props.historyByTarget.get(targetID()); + const buckets = () => history()?.buckets ?? []; + const source = () => + getProbeSourceChipLabel( + props.probeAgentOptions ?? [], + availability()?.probeAgentId, + ) ?? 'Local Pulse'; + const endpoint = () => { + const current = availability(); + return current + ? getAvailabilityProbeEndpointLabel(current) || resource.name + : resource.name; + }; + const checked = () => + formatRelativeTime(availability()?.lastChecked, { + compact: true, + emptyText: 'Not checked', + }); + const latency = () => { + const value = availability()?.latencyMillis; + return typeof value === 'number' && Number.isFinite(value) && value > 0 + ? `${value.toLocaleString()} ms` + : undefined; + }; + + return ( + + ); + }} + +
+ + + {(resource) => ( + setSelectedResource(undefined)} + /> + )} + +
+ ); +}; + +export default AvailabilityFleetView; diff --git a/frontend-modern/src/features/standalone/StandalonePageSurface.tsx b/frontend-modern/src/features/standalone/StandalonePageSurface.tsx index 0b2a89286..04f2dab88 100644 --- a/frontend-modern/src/features/standalone/StandalonePageSurface.tsx +++ b/frontend-modern/src/features/standalone/StandalonePageSurface.tsx @@ -33,7 +33,7 @@ import { import { updateStore } from '@/stores/updates'; import { formatRelativeTime } from '@/utils/format'; import { buildProbeAgentOptions } from '@/utils/availabilityProbeAgents'; -import { AvailabilityChecksTable } from './AvailabilityChecksTable'; +import { AvailabilityChecksTable, type AvailabilityChecksView } from './AvailabilityChecksTable'; import { AgentsMachinesTable } from './AgentsMachinesTable'; import { collectHostIdentityConflictHosts } from './hostIdentityConflict'; import { HostIdentityConflictNotice } from './HostIdentityConflictNotice'; @@ -169,6 +169,15 @@ export function StandalonePageSurface() { { replace: true }, ); }; + const availabilityView = createMemo(() => + searchParams[STANDALONE_QUERY_PARAMS.view] === 'fleet' ? 'fleet' : 'table', + ); + const setAvailabilityView = (view: AvailabilityChecksView) => { + setSearchParams( + { [STANDALONE_QUERY_PARAMS.view]: view === 'table' ? null : view }, + { replace: true }, + ); + }; const availabilityPosture = createMemo(() => buildStandalonePostureSummary(model().availabilityChecks), ); @@ -265,6 +274,13 @@ export function StandalonePageSurface() { ({ + ResourceDetailDrawer: (props: { resource: Resource; onClose?: () => void }) => ( +
+ +
+ ), +})); + +const resource = (index: number): Resource => + ({ + id: `availability:target-${index}`, + name: `Service ${index}`, + displayName: `Service ${index}`, + type: 'network-endpoint', + platformId: `target-${index}`, + platformType: 'availability', + sourceType: 'api', + sources: ['availability'], + status: index % 5 === 0 ? 'offline' : 'online', + lastSeen: Date.parse('2026-08-30T11:59:00Z'), + availability: { + targetId: `target-${index}`, + protocol: index % 2 === 0 ? 'https' : 'tcp', + address: `service-${index}.lab.local`, + enabled: true, + available: index % 5 !== 0, + outcome: index % 5 === 0 ? 'unreachable' : 'reachable', + latencyMillis: index % 5 === 0 ? undefined : 10 + index, + lastChecked: '2026-08-30T11:59:00Z', + pollIntervalSeconds: 60, + }, + }) as Resource; + +const history = (index: number): AvailabilityHistoryTarget => ({ + targetId: `target-${index}`, + summary: { + reachableSeconds: 3600, + unreachableSeconds: index % 5 === 0 ? 600 : 0, + indeterminateSeconds: 300, + unknownSeconds: 82_500, + coveragePercent: 4.51, + availabilityPercent: index % 5 === 0 ? 85.71 : 100, + reachableLatencyMillis: { average: 15 + index, min: 8, max: 42 }, + }, + buckets: [ + { + start: '2026-08-30T08:00:00Z', + end: '2026-08-30T09:00:00Z', + reachableSeconds: 3600, + unreachableSeconds: 0, + indeterminateSeconds: 0, + unknownSeconds: 0, + latencyMillis: { average: 12, min: 8, max: 16 }, + }, + { + start: '2026-08-30T09:00:00Z', + end: '2026-08-30T10:00:00Z', + reachableSeconds: 0, + unreachableSeconds: 0, + indeterminateSeconds: 3600, + unknownSeconds: 0, + }, + { + start: '2026-08-30T10:00:00Z', + end: '2026-08-30T11:00:00Z', + reachableSeconds: 0, + unreachableSeconds: index % 5 === 0 ? 3600 : 0, + indeterminateSeconds: 0, + unknownSeconds: index % 5 === 0 ? 0 : 3600, + }, + { + start: '2026-08-30T11:00:00Z', + end: '2026-08-30T12:00:00Z', + reachableSeconds: 3600, + unreachableSeconds: 0, + indeterminateSeconds: 0, + unknownSeconds: 0, + latencyMillis: { average: 18, min: 14, max: 22 }, + }, + ], + revisionBoundaries: [{ revision: 2, at: '2026-08-30T10:30:00Z' }], +}); + +afterEach(cleanup); + +describe('AvailabilityFleetView', () => { + it('renders fifty keyboard-accessible attention tiles with non-color history labels', () => { + const resources = Array.from({ length: 50 }, (_, index) => resource(index)); + const historyByTarget = new Map( + resources.map((_item, index) => [`target-${index}`, history(index)]), + ); + + const view = render(() => ( + + )); + + expect(view.container.querySelectorAll('[data-availability-fleet-tile]')).toHaveLength(50); + expect(screen.getByText('Reachable')).toBeInTheDocument(); + expect(screen.getByText('Unreachable')).toBeInTheDocument(); + expect(screen.getByText('Indeterminate')).toBeInTheDocument(); + expect(screen.getByText('Unknown')).toBeInTheDocument(); + expect( + view.container.querySelector('[data-testid="availability-state-strip"]'), + ).toHaveAttribute('aria-label', expect.stringContaining('indeterminate')); + expect( + view.container.querySelectorAll('[data-testid="availability-latency-line"]').length, + ).toBeGreaterThan(0); + expect(screen.getAllByText(/Insufficient coverage/)).toHaveLength(50); + + const tile = screen.getByRole('button', { name: 'Open details for Service 0' }); + expect(tile).toHaveAttribute('type', 'button'); + fireEvent.click(tile); + expect(screen.getByTestId('resource-detail-drawer')).toHaveAttribute( + 'data-resource-id', + 'availability:target-0', + ); + }); + + it('keeps current health visible when history is unavailable', () => { + render(() => ( + + )); + + expect( + screen.getByText('History unavailable. Current health is unchanged.'), + ).toBeInTheDocument(); + expect(screen.getByText('History unavailable')).toBeInTheDocument(); + expect(screen.getByText('Reachable')).toBeInTheDocument(); + }); +}); diff --git a/frontend-modern/src/routing/__tests__/resourceLinks.test.ts b/frontend-modern/src/routing/__tests__/resourceLinks.test.ts index 8d617ae21..a79c26e11 100644 --- a/frontend-modern/src/routing/__tests__/resourceLinks.test.ts +++ b/frontend-modern/src/routing/__tests__/resourceLinks.test.ts @@ -187,7 +187,7 @@ describe('resource link routing contract', () => { it('builds canonical Machines, container runtime, Kubernetes, TrueNAS, and vSphere tab paths', () => { expect(STANDALONE_PATH).toBe('/standalone'); expect(STANDALONE_DEFAULT_TAB).toBe('machines'); - expect(STANDALONE_QUERY_PARAMS).toEqual({ query: 'q', status: 'status' }); + expect(STANDALONE_QUERY_PARAMS).toEqual({ query: 'q', status: 'status', view: 'view' }); expect(buildStandalonePath()).toBe('/standalone/machines'); expect(buildStandalonePath('')).toBe('/standalone'); diff --git a/frontend-modern/src/routing/resourceLinks.ts b/frontend-modern/src/routing/resourceLinks.ts index 53066be67..d9e8e1310 100644 --- a/frontend-modern/src/routing/resourceLinks.ts +++ b/frontend-modern/src/routing/resourceLinks.ts @@ -21,6 +21,7 @@ export const STANDALONE_DEFAULT_TAB = 'machines'; export const STANDALONE_QUERY_PARAMS = { query: 'q', status: 'status', + view: 'view', } as const; export const PROXMOX_PATH = '/proxmox'; export const PROXMOX_DEFAULT_TAB = 'overview'; diff --git a/internal/api/availability_handlers.go b/internal/api/availability_handlers.go index c1e4e1f61..80d8025d2 100644 --- a/internal/api/availability_handlers.go +++ b/internal/api/availability_handlers.go @@ -145,6 +145,7 @@ func (h *AvailabilityHandlers) HandleAdd(w http.ResponseWriter, r *http.Request) return } target = config.NormalizeAvailabilityTarget(target) + target.ConfigRevision = 1 if err := target.Validate(); err != nil { writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) return @@ -209,12 +210,17 @@ func (h *AvailabilityHandlers) HandleUpdate(w http.ResponseWriter, r *http.Reque return } - target, ok := decodeAvailabilityTargetRequest(w, r, targets[index]) + previous := config.NormalizeAvailabilityTarget(targets[index]) + target, ok := decodeAvailabilityTargetRequest(w, r, previous) if !ok { return } target.ID = targetID target = config.NormalizeAvailabilityTarget(target) + target.ConfigRevision = previous.ConfigRevision + if config.AvailabilityExecutionConfigChanged(previous, target) { + target.ConfigRevision++ + } if err := target.Validate(); err != nil { writeErrorResponse(w, http.StatusBadRequest, "validation_error", err.Error(), nil) return @@ -267,6 +273,11 @@ func (h *AvailabilityHandlers) HandleDelete(w http.ResponseWriter, r *http.Reque writeErrorResponse(w, http.StatusInternalServerError, "availability_save_failed", "Failed to save availability targets", map[string]string{"error": err.Error()}) return } + if monitor := h.monitorForRequest(r.Context()); monitor != nil { + if store := monitor.GetMetricsStore(); store != nil { + store.DeleteAvailabilityTargetHistory(targetID) + } + } h.refreshMonitor(r.Context()) writeJSON(w, http.StatusOK, map[string]any{"success": true, "id": targetID}) } diff --git a/internal/api/availability_handlers_test.go b/internal/api/availability_handlers_test.go index b8c715784..fc442089e 100644 --- a/internal/api/availability_handlers_test.go +++ b/internal/api/availability_handlers_test.go @@ -43,6 +43,9 @@ func TestAvailabilityHandlersCRUDPersistsTargets(t *testing.T) { if created.ID == "" { t.Fatal("created ID is empty") } + if created.ConfigRevision != 1 { + t.Fatalf("created config revision = %d, want 1", created.ConfigRevision) + } updated := created updated.Enabled = false @@ -61,6 +64,27 @@ func TestAvailabilityHandlersCRUDPersistsTargets(t *testing.T) { if len(loaded) != 1 || loaded[0].Enabled { t.Fatalf("loaded targets = %+v, want one paused target", loaded) } + if loaded[0].ConfigRevision != 1 { + t.Fatalf("non-execution edit revision = %d, want 1", loaded[0].ConfigRevision) + } + + executionEdit := loaded[0] + executionEdit.Address = "gateway-2.local" + executionEdit.ConfigRevision = 99 + updateBody = availabilityRequestBody(t, executionEdit) + updateReq = httptest.NewRequest(http.MethodPut, "/api/availability-targets/"+created.ID, updateBody) + updateRec = httptest.NewRecorder() + handler.HandleUpdate(updateRec, updateReq) + if updateRec.Code != http.StatusOK { + t.Fatalf("execution HandleUpdate status = %d, body=%s", updateRec.Code, updateRec.Body.String()) + } + loaded, err = persistence.LoadAvailabilityTargets() + if err != nil { + t.Fatalf("LoadAvailabilityTargets() after execution edit error = %v", err) + } + if loaded[0].ConfigRevision != 2 { + t.Fatalf("server-authored execution edit revision = %d, want 2", loaded[0].ConfigRevision) + } listReq := httptest.NewRequest(http.MethodGet, "/api/availability-targets", nil) listRec := httptest.NewRecorder() diff --git a/internal/api/availability_history_handlers.go b/internal/api/availability_history_handlers.go new file mode 100644 index 000000000..d49e3d250 --- /dev/null +++ b/internal/api/availability_history_handlers.go @@ -0,0 +1,257 @@ +package api + +import ( + "encoding/json" + "math" + "net/http" + "strings" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/internal/mock" + pkgmetrics "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" +) + +const ( + availabilityHistoryMaxTargets = 200 + availabilityHistoryMaxBuckets = 120 +) + +var availabilityHistoryRanges = map[string]time.Duration{ + "1h": time.Hour, + "6h": 6 * time.Hour, + "12h": 12 * time.Hour, + "24h": 24 * time.Hour, + "7d": 7 * 24 * time.Hour, + "14d": 14 * 24 * time.Hour, + "30d": 30 * 24 * time.Hour, + "90d": 90 * 24 * time.Hour, + "365d": 365 * 24 * time.Hour, +} + +type availabilityHistoryRequest struct { + TargetIDs []string `json:"targetIds"` + Range string `json:"range,omitempty"` +} + +type availabilityHistoryTargetError struct { + Code string `json:"code"` + Message string `json:"message"` +} + +type availabilityHistoryTargetResponse struct { + TargetID string `json:"targetId"` + Summary *pkgmetrics.AvailabilityHistorySummary `json:"summary,omitempty"` + Buckets []pkgmetrics.AvailabilityHistoryBucket `json:"buckets,omitempty"` + RevisionBoundaries []pkgmetrics.AvailabilityRevisionBoundary `json:"revisionBoundaries,omitempty"` + Error *availabilityHistoryTargetError `json:"error,omitempty"` +} + +type availabilityHistoryResponse struct { + Start time.Time `json:"start"` + End time.Time `json:"end"` + Targets []availabilityHistoryTargetResponse `json:"targets"` +} + +func (r *Router) handleAvailabilityHistory(w http.ResponseWriter, req *http.Request) { + if req.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + + req.Body = http.MaxBytesReader(w, req.Body, 32*1024) + defer req.Body.Close() + var body availabilityHistoryRequest + decoder := json.NewDecoder(req.Body) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&body); err != nil { + writeErrorResponse(w, http.StatusBadRequest, "invalid_request", "Invalid availability history request", nil) + return + } + body.TargetIDs = normalizeAvailabilityHistoryIDs(body.TargetIDs) + if len(body.TargetIDs) == 0 { + writeErrorResponse(w, http.StatusBadRequest, "missing_target_ids", "At least one availability target ID is required", nil) + return + } + if len(body.TargetIDs) > availabilityHistoryMaxTargets { + writeErrorResponse(w, http.StatusBadRequest, "too_many_target_ids", "Availability history accepts at most 200 unique target IDs", nil) + return + } + + rangeName := strings.ToLower(strings.TrimSpace(body.Range)) + if rangeName == "" { + rangeName = "24h" + } + duration, ok := availabilityHistoryRanges[rangeName] + if !ok { + writeErrorResponse(w, http.StatusBadRequest, "invalid_range", "Unsupported availability history range", nil) + return + } + maxHistoryDays := freeHistoryDaysDefault + if r.licenseHandlers != nil { + if service := r.licenseHandlers.Service(req.Context()); service != nil { + status := service.Status() + if status.Valid { + maxHistoryDays = tierHistoryDaysFromLicensing(status.Tier) + } + } + } + if duration > time.Duration(maxHistoryDays)*24*time.Hour { + WriteLicenseRequired(w, featureLongTermMetricsValue, "Extended availability history requires a higher-tier Pulse license") + return + } + + configured := make(map[string]config.AvailabilityTarget) + if mock.IsMockEnabled() { + for _, response := range mockAvailabilityTargetResponses() { + target := config.NormalizeAvailabilityTarget(response.AvailabilityTarget) + configured[target.ID] = target + } + } else { + persistence := r.persistenceForOrg(req.Context()) + if persistence == nil { + writeErrorResponse(w, http.StatusInternalServerError, "availability_unavailable", "Availability target persistence is unavailable", nil) + return + } + targets, err := persistence.LoadAvailabilityTargets() + if err != nil { + writeErrorResponse(w, http.StatusInternalServerError, "availability_load_failed", "Failed to load availability targets", nil) + return + } + for _, target := range targets { + target = config.NormalizeAvailabilityTarget(target) + configured[target.ID] = target + } + } + + queryIDs := make([]string, 0, len(body.TargetIDs)) + for _, targetID := range body.TargetIDs { + if _, exists := configured[targetID]; exists { + queryIDs = append(queryIDs, targetID) + } + } + end := time.Now().UTC().Truncate(time.Minute) + start := end.Add(-duration) + + var results map[string]pkgmetrics.AvailabilityHistoryTarget + if mock.IsMockEnabled() { + results = mockAvailabilityHistory(queryIDs, start, end) + } else { + monitor := r.getTenantMonitor(req.Context()) + if monitor == nil || monitor.GetMetricsStore() == nil { + writeErrorResponse(w, http.StatusServiceUnavailable, "availability_history_unavailable", "Availability history is unavailable", nil) + return + } + var err error + results, err = monitor.GetMetricsStore().QueryAvailabilityHistory(queryIDs, start, end, availabilityHistoryMaxBuckets) + if err != nil { + writeErrorResponse(w, http.StatusInternalServerError, "availability_history_failed", "Failed to load availability history", nil) + return + } + } + + response := availabilityHistoryResponse{Start: start, End: end, Targets: make([]availabilityHistoryTargetResponse, 0, len(body.TargetIDs))} + for _, targetID := range body.TargetIDs { + if _, exists := configured[targetID]; !exists { + response.Targets = append(response.Targets, availabilityHistoryTargetResponse{ + TargetID: targetID, + Error: &availabilityHistoryTargetError{Code: "not_found", Message: "Availability target not found"}, + }) + continue + } + history := results[targetID] + response.Targets = append(response.Targets, availabilityHistoryTargetResponse{ + TargetID: targetID, + Summary: &history.Summary, + Buckets: history.Buckets, + RevisionBoundaries: history.RevisionBoundaries, + }) + } + writeJSON(w, http.StatusOK, response) +} + +func mockAvailabilityHistory(targetIDs []string, start, end time.Time) map[string]pkgmetrics.AvailabilityHistoryTarget { + const bucketCount = 120 + result := make(map[string]pkgmetrics.AvailabilityHistoryTarget, len(targetIDs)) + step := end.Sub(start) / bucketCount + for targetIndex, targetID := range targetIDs { + buckets := make([]pkgmetrics.AvailabilityHistoryBucket, 0, bucketCount) + summary := pkgmetrics.AvailabilityHistorySummary{} + latencyCount, latencySum := int64(0), float64(0) + latencyMin, latencyMax := int64(0), int64(0) + for bucketIndex := 0; bucketIndex < bucketCount; bucketIndex++ { + bucketStart := start.Add(time.Duration(bucketIndex) * step) + bucketEnd := bucketStart.Add(step) + seconds := bucketEnd.Sub(bucketStart).Seconds() + bucket := pkgmetrics.AvailabilityHistoryBucket{Start: bucketStart, End: bucketEnd} + pattern := (bucketIndex + targetIndex*7) % 53 + switch { + case pattern == 0 || pattern == 1: + bucket.UnknownSeconds = seconds + summary.UnknownSeconds += seconds + case pattern == 9: + bucket.IndeterminateSeconds = seconds + summary.IndeterminateSeconds += seconds + case (targetIndex%5 == 0 && pattern >= 30 && pattern <= 33) || pattern == 21: + bucket.UnreachableSeconds = seconds + summary.UnreachableSeconds += seconds + default: + bucket.ReachableSeconds = seconds + summary.ReachableSeconds += seconds + latency := int64(8 + (bucketIndex*3+targetIndex*5)%44) + bucket.LatencyMillis = &pkgmetrics.AvailabilityLatencySummary{Average: float64(latency), Min: latency, Max: latency} + latencyCount++ + latencySum += float64(latency) + if latencyMin == 0 || latency < latencyMin { + latencyMin = latency + } + if latency > latencyMax { + latencyMax = latency + } + } + buckets = append(buckets, bucket) + } + known := summary.ReachableSeconds + summary.UnreachableSeconds + summary.IndeterminateSeconds + summary.CoveragePercent = math.Round(known*10000/end.Sub(start).Seconds()) / 100 + determinate := summary.ReachableSeconds + summary.UnreachableSeconds + if determinate > 0 { + availability := math.Round(summary.ReachableSeconds*10000/determinate) / 100 + summary.AvailabilityPercent = &availability + } + if latencyCount > 0 { + summary.ReachableLatencyMillis = &pkgmetrics.AvailabilityLatencySummary{ + Average: math.Round(latencySum/float64(latencyCount)*100) / 100, + Min: latencyMin, + Max: latencyMax, + } + } + boundaries := []pkgmetrics.AvailabilityRevisionBoundary{} + if targetIndex%9 == 0 { + boundaries = append(boundaries, pkgmetrics.AvailabilityRevisionBoundary{Revision: 2, At: start.Add(end.Sub(start) / 2)}) + } + result[targetID] = pkgmetrics.AvailabilityHistoryTarget{ + TargetID: targetID, + Summary: summary, + Buckets: buckets, + RevisionBoundaries: boundaries, + } + } + return result +} + +func normalizeAvailabilityHistoryIDs(targetIDs []string) []string { + seen := make(map[string]struct{}, len(targetIDs)) + ids := make([]string, 0, len(targetIDs)) + for _, targetID := range targetIDs { + targetID = strings.TrimSpace(targetID) + if targetID == "" { + continue + } + if _, exists := seen[targetID]; exists { + continue + } + seen[targetID] = struct{}{} + ids = append(ids, targetID) + } + return ids +} diff --git a/internal/api/availability_history_handlers_test.go b/internal/api/availability_history_handlers_test.go new file mode 100644 index 000000000..a5089ae49 --- /dev/null +++ b/internal/api/availability_history_handlers_test.go @@ -0,0 +1,95 @@ +package api + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" +) + +func TestHandleAvailabilityHistoryReturnsBatchResultsAndPerTargetNotFound(t *testing.T) { + monitor, _, _ := newTestMonitor(t) + store, err := metrics.NewStore(metrics.DefaultConfig(t.TempDir())) + if err != nil { + t.Fatal(err) + } + defer store.Close() + setUnexportedField(t, monitor, "metricsStore", store) + + persistence := config.NewConfigPersistence(t.TempDir()) + target := config.NormalizeAvailabilityTarget(config.AvailabilityTarget{ + ID: "target-one", Name: "Gateway", Address: "gateway.local", TargetKind: config.AvailabilityTargetService, + Protocol: config.AvailabilityProbeHTTPS, Enabled: true, + }) + if err := persistence.SaveAvailabilityTargets([]config.AvailabilityTarget{target}); err != nil { + t.Fatal(err) + } + latency := int64(24) + at := time.Now().UTC().Truncate(time.Minute).Add(-10 * time.Minute) + if err := store.WriteAvailabilityObservationSync(metrics.AvailabilityObservation{ + ObservationID: "api-observation", TargetID: target.ID, ConfigRevision: target.ConfigRevision, + Outcome: metrics.AvailabilityReachable, ObservedAt: at, TimelineAt: at, IngestedAt: at, + ValidFor: 5 * time.Minute, ExecutionSource: metrics.AvailabilitySourceLocal, LatencyMillis: &latency, + }); err != nil { + t.Fatal(err) + } + + router := &Router{monitor: monitor, persistence: persistence} + body, _ := json.Marshal(availabilityHistoryRequest{TargetIDs: []string{target.ID, "missing"}, Range: "24h"}) + req := httptest.NewRequest(http.MethodPost, "/api/availability-history", bytes.NewReader(body)) + rec := httptest.NewRecorder() + router.handleAvailabilityHistory(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, body=%s", rec.Code, rec.Body.String()) + } + var response availabilityHistoryResponse + if err := json.NewDecoder(rec.Body).Decode(&response); err != nil { + t.Fatal(err) + } + if len(response.Targets) != 2 { + t.Fatalf("targets = %+v", response.Targets) + } + if response.Targets[0].Summary == nil || response.Targets[0].Summary.ReachableSeconds != 300 { + t.Fatalf("known target = %+v", response.Targets[0]) + } + if len(response.Targets[0].Buckets) == 0 || len(response.Targets[0].Buckets) > availabilityHistoryMaxBuckets { + t.Fatalf("bucket count = %d", len(response.Targets[0].Buckets)) + } + if response.Targets[1].Error == nil || response.Targets[1].Error.Code != "not_found" { + t.Fatalf("missing target = %+v", response.Targets[1]) + } +} + +func TestHandleAvailabilityHistoryEnforcesMethodRangeAndUniqueTargetBound(t *testing.T) { + router := &Router{} + + rec := httptest.NewRecorder() + router.handleAvailabilityHistory(rec, httptest.NewRequest(http.MethodGet, "/api/availability-history", nil)) + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("GET status = %d", rec.Code) + } + + body, _ := json.Marshal(availabilityHistoryRequest{TargetIDs: []string{"one"}, Range: "14d"}) + rec = httptest.NewRecorder() + router.handleAvailabilityHistory(rec, httptest.NewRequest(http.MethodPost, "/api/availability-history", bytes.NewReader(body))) + if rec.Code != http.StatusPaymentRequired { + t.Fatalf("14d free status = %d, body=%s", rec.Code, rec.Body.String()) + } + + ids := make([]string, availabilityHistoryMaxTargets+1) + for index := range ids { + ids[index] = fmt.Sprintf("target-%d", index) + } + body, _ = json.Marshal(availabilityHistoryRequest{TargetIDs: ids, Range: "24h"}) + rec = httptest.NewRecorder() + router.handleAvailabilityHistory(rec, httptest.NewRequest(http.MethodPost, "/api/availability-history", bytes.NewReader(body))) + if rec.Code != http.StatusBadRequest { + t.Fatalf("201 target status = %d, body=%s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/api/route_inventory_test.go b/internal/api/route_inventory_test.go index 76b8e4ead..bee1ed84b 100644 --- a/internal/api/route_inventory_test.go +++ b/internal/api/route_inventory_test.go @@ -408,6 +408,7 @@ var allRouteAllowlist = []string{ "/api/charts/workloads-summary", "/api/metrics-store/stats", "/api/metrics-store/history", + "/api/availability-history", "/api/diagnostics", "/api/diagnostics/docker/prepare-token", "/api/config", diff --git a/internal/api/router_routes_monitoring.go b/internal/api/router_routes_monitoring.go index 90bd9dd8f..51d61826a 100644 --- a/internal/api/router_routes_monitoring.go +++ b/internal/api/router_routes_monitoring.go @@ -25,6 +25,7 @@ func (r *Router) registerMonitoringResourceRoutes( r.mux.HandleFunc("/api/charts/workloads-summary", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleWorkloadsSummaryCharts))) r.mux.HandleFunc("/api/metrics-store/stats", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleMetricsStoreStats))) r.mux.HandleFunc("/api/metrics-store/history", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleMetricsHistory))) + r.mux.HandleFunc("/api/availability-history", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.handleAvailabilityHistory))) r.mux.HandleFunc("/api/recovery/points", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.recoveryHandlers.HandleListPoints))) r.mux.HandleFunc("/api/recovery/series", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.recoveryHandlers.HandleListSeries))) r.mux.HandleFunc("/api/recovery/facets", RequireAuth(r.config, RequireScope(config.ScopeMonitoringRead, r.recoveryHandlers.HandleListFacets))) diff --git a/internal/api/security_regression_test.go b/internal/api/security_regression_test.go index 397a1adb6..807ce89ca 100644 --- a/internal/api/security_regression_test.go +++ b/internal/api/security_regression_test.go @@ -3230,6 +3230,7 @@ func TestMonitoringReadEndpointsRequireMonitoringReadScope(t *testing.T) { "/api/charts/storage-summary", "/api/metrics-store/stats", "/api/metrics-store/history", + "/api/availability-history", "/api/guests/metadata", "/api/guests/metadata/guest-1", "/api/docker/metadata", diff --git a/internal/config/availability.go b/internal/config/availability.go index 16bc206b5..06e131a94 100644 --- a/internal/config/availability.go +++ b/internal/config/availability.go @@ -51,6 +51,7 @@ const ( // lightweight availability probe. type AvailabilityTarget struct { ID string `json:"id"` + ConfigRevision int64 `json:"configRevision"` Name string `json:"name"` TargetKind AvailabilityTargetKind `json:"targetKind,omitempty"` Address string `json:"address"` @@ -80,6 +81,7 @@ type AvailabilityTarget struct { func NewAvailabilityTarget() AvailabilityTarget { return AvailabilityTarget{ ID: uuid.NewString(), + ConfigRevision: 1, TargetKind: AvailabilityTargetService, Protocol: AvailabilityProbeICMP, Enabled: true, @@ -96,6 +98,9 @@ func (t *AvailabilityTarget) ApplyDefaults() { if strings.TrimSpace(t.ID) == "" { t.ID = uuid.NewString() } + if t.ConfigRevision <= 0 { + t.ConfigRevision = 1 + } if strings.TrimSpace(string(t.Protocol)) == "" { t.Protocol = AvailabilityProbeICMP } else { @@ -142,6 +147,24 @@ func (t AvailabilityTarget) EffectiveFailureThreshold() int { return DefaultAvailabilityFailureThreshold } +// AvailabilityExecutionConfigChanged reports whether an edit changes what is +// executed or where it executes. Display, correlation, alert-threshold, and +// certificate-presentation edits intentionally stay within the same revision. +func AvailabilityExecutionConfigChanged(previous, next AvailabilityTarget) bool { + previous = NormalizeAvailabilityTarget(previous) + next = NormalizeAvailabilityTarget(next) + return previous.Address != next.Address || + previous.Protocol != next.Protocol || + previous.Port != next.Port || + previous.Path != next.Path || + previous.UDPMode != next.UDPMode || + previous.UDPRequest != next.UDPRequest || + previous.UDPExpected != next.UDPExpected || + previous.EffectiveTimeoutMillis() != next.EffectiveTimeoutMillis() || + previous.EffectivePollIntervalSecs() != next.EffectivePollIntervalSecs() || + previous.ProbeAgentID != next.ProbeAgentID +} + func (t AvailabilityTarget) CertificateMonitoringEnabled() bool { return normalizeAvailabilityProbeProtocol(t.Protocol) == AvailabilityProbeHTTPS && !t.CertificateMonitoringDisabled } diff --git a/internal/config/availability_test.go b/internal/config/availability_test.go index 5f688c377..6c6609c4a 100644 --- a/internal/config/availability_test.go +++ b/internal/config/availability_test.go @@ -29,6 +29,40 @@ func TestNormalizeAvailabilityTargetPreservesHTTPAddress(t *testing.T) { } } +func TestAvailabilityExecutionConfigChangedTracksOnlyExecutionDefiningFields(t *testing.T) { + base := NormalizeAvailabilityTarget(AvailabilityTarget{ + ID: "target-1", Name: "Gateway", Address: "gateway.local", + Protocol: AvailabilityProbeTCP, Port: 443, Enabled: true, + PollIntervalSecs: 60, TimeoutMillis: 1500, ProbeAgentID: "agent-1", + }) + + presentationOnly := base + presentationOnly.Name = "Primary gateway" + presentationOnly.LinkedResourceID = "node-1" + presentationOnly.FailureThreshold = 5 + if AvailabilityExecutionConfigChanged(base, presentationOnly) { + t.Fatal("presentation and alert-only edits changed the execution revision") + } + + for name, mutate := range map[string]func(*AvailabilityTarget){ + "address": func(target *AvailabilityTarget) { target.Address = "gateway-2.local" }, + "protocol": func(target *AvailabilityTarget) { target.Protocol = AvailabilityProbeICMP }, + "port": func(target *AvailabilityTarget) { target.Port = 8443 }, + "path": func(target *AvailabilityTarget) { target.Path = "/health" }, + "timeout": func(target *AvailabilityTarget) { target.TimeoutMillis = 2500 }, + "poll interval": func(target *AvailabilityTarget) { target.PollIntervalSecs = 120 }, + "probe agent": func(target *AvailabilityTarget) { target.ProbeAgentID = "agent-2" }, + } { + t.Run(name, func(t *testing.T) { + next := base + mutate(&next) + if !AvailabilityExecutionConfigChanged(base, next) { + t.Fatalf("%s edit did not change the execution revision", name) + } + }) + } +} + func TestNormalizeAvailabilityTargetReducesICMPAddressToHost(t *testing.T) { target := NormalizeAvailabilityTarget(AvailabilityTarget{ Address: " https://device.local:8443/status ", diff --git a/internal/hostagent/availability.go b/internal/hostagent/availability.go index 320ba681f..a0b0a8945 100644 --- a/internal/hostagent/availability.go +++ b/internal/hostagent/availability.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/utils" @@ -242,11 +243,13 @@ func (m *availabilityProbeModule) check(ctx context.Context, target config.Avail } result := agentshost.AvailabilityProbeResult{ - TargetID: target.ID, - Outcome: string(probeResult.Outcome), - LatencyMillis: latency.Milliseconds(), - CheckedAt: m.now().UTC(), - Certificate: probeResult.Certificate.Clone(), + ObservationID: uuid.NewString(), + TargetID: target.ID, + ConfigRevision: target.ConfigRevision, + Outcome: string(probeResult.Outcome), + LatencyMillis: latency.Milliseconds(), + CheckedAt: m.now().UTC(), + Certificate: probeResult.Certificate.Clone(), } if err != nil { message := strings.TrimSpace(err.Error()) diff --git a/internal/monitoring/availability_poller.go b/internal/monitoring/availability_poller.go index 311381e6b..5ad3ad825 100644 --- a/internal/monitoring/availability_poller.go +++ b/internal/monitoring/availability_poller.go @@ -9,12 +9,15 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/rcourtman/pulse-go-rewrite/internal/availabilityprobe" "github.com/rcourtman/pulse-go-rewrite/internal/config" "github.com/rcourtman/pulse-go-rewrite/internal/operationaltrust" "github.com/rcourtman/pulse-go-rewrite/internal/storagehealth" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" + pkgmetrics "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" "github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil" + "github.com/rs/zerolog/log" ) type tlsCert = tlsutil.CertificateObservation @@ -329,7 +332,7 @@ func (m *Monitor) pollAvailabilityTarget(ctx context.Context, target config.Avai result, err := ProbeAvailabilityTargetDetailedResult(ctx, target) latency := time.Since(start) checkedAt := time.Now().UTC() - m.applyAvailabilityObservation(target, checkedAt, latency, result.Outcome, err, result.Certificate, "", time.Time{}) + m.applyAvailabilityObservation(target, uuid.NewString(), checkedAt, latency, result.Outcome, err, result.Certificate, "", time.Time{}) m.updateResourceStore(m.GetState()) } @@ -338,6 +341,7 @@ func (m *Monitor) pollAvailabilityTarget(ctx context.Context, target config.Avai // accounting, connection health, and task bookkeeping. func (m *Monitor) applyAvailabilityObservation( target config.AvailabilityTarget, + observationID string, checkedAt time.Time, latency time.Duration, outcome AvailabilityProbeOutcome, @@ -347,6 +351,7 @@ func (m *Monitor) applyAvailabilityObservation( probeReportReceivedAt time.Time, ) { m.setAvailabilityStatusWithCertificate(target, checkedAt, latency, outcome, probeErr, certificate, probeAgentID, probeReportReceivedAt) + m.recordAvailabilityHistory(target, observationID, checkedAt, latency, outcome, probeErr, probeAgentID, probeReportReceivedAt) if probeErr == nil { if m.stalenessTracker != nil { @@ -362,6 +367,53 @@ func (m *Monitor) applyAvailabilityObservation( m.recordTaskResult(InstanceTypeAvailability, target.ID, nil) } +func (m *Monitor) recordAvailabilityHistory( + target config.AvailabilityTarget, + observationID string, + checkedAt time.Time, + latency time.Duration, + outcome AvailabilityProbeOutcome, + probeErr error, + probeAgentID string, + probeReportReceivedAt time.Time, +) { + if m == nil || m.metricsStore == nil { + return + } + ingestedAt := time.Now().UTC() + timelineAt := checkedAt.UTC() + source := pkgmetrics.AvailabilitySourceLocal + if strings.TrimSpace(probeAgentID) != "" { + source = pkgmetrics.AvailabilitySourceAssignedAgent + if !probeReportReceivedAt.IsZero() { + timelineAt = probeReportReceivedAt.UTC() + ingestedAt = timelineAt + } + } + var latencyMillis *int64 + if probeErr == nil && outcome == AvailabilityProbeReachable { + value := latency.Milliseconds() + if value == 0 { + value = 1 + } + latencyMillis = &value + } + if err := m.metricsStore.WriteAvailabilityObservationBounded(pkgmetrics.AvailabilityObservation{ + ObservationID: observationID, + TargetID: target.ID, + ConfigRevision: target.ConfigRevision, + Outcome: pkgmetrics.AvailabilityOutcome(outcome), + ObservedAt: checkedAt.UTC(), + TimelineAt: timelineAt, + IngestedAt: ingestedAt, + ValidFor: availabilityProbeStaleWindow(target), + ExecutionSource: source, + LatencyMillis: latencyMillis, + }); err != nil { + log.Warn().Err(err).Str("target_id", target.ID).Msg("Dropping invalid availability history observation") + } +} + func (m *Monitor) setAvailabilityStatus( target config.AvailabilityTarget, checkedAt time.Time, diff --git a/internal/monitoring/availability_probe_agent.go b/internal/monitoring/availability_probe_agent.go index 3ba57993e..a7aa2e76d 100644 --- a/internal/monitoring/availability_probe_agent.go +++ b/internal/monitoring/availability_probe_agent.go @@ -1,7 +1,9 @@ package monitoring import ( + "crypto/sha256" "errors" + "fmt" "strings" "time" @@ -24,12 +26,14 @@ const availabilityProbeStaleError = "no recent report from probe agent" // ProbeAvailabilityResult is one availability observation reported by a remote // host agent that owns the target's execution. type ProbeAvailabilityResult struct { - TargetID string - Outcome availabilityprobe.Outcome - LatencyMillis int64 - CheckedAt time.Time - Error string - Certificate *tlsutil.CertificateObservation + ObservationID string + TargetID string + ConfigRevision int64 + Outcome availabilityprobe.Outcome + LatencyMillis int64 + CheckedAt time.Time + Error string + Certificate *tlsutil.CertificateObservation } // availabilityProbeAssignmentTracker provides a grace reference for a newly @@ -57,12 +61,14 @@ func probeAvailabilityResultsFromReport(reported []agentshost.AvailabilityProbeR outcome = availabilityprobe.OutcomeIndeterminate } results = append(results, ProbeAvailabilityResult{ - TargetID: strings.TrimSpace(entry.TargetID), - Outcome: outcome, - LatencyMillis: entry.LatencyMillis, - CheckedAt: entry.CheckedAt, - Error: strings.TrimSpace(entry.Error), - Certificate: entry.Certificate.Clone(), + ObservationID: strings.TrimSpace(entry.ObservationID), + TargetID: strings.TrimSpace(entry.TargetID), + ConfigRevision: entry.ConfigRevision, + Outcome: outcome, + LatencyMillis: entry.LatencyMillis, + CheckedAt: entry.CheckedAt, + Error: strings.TrimSpace(entry.Error), + Certificate: entry.Certificate.Clone(), }) } return results @@ -128,6 +134,15 @@ func (m *Monitor) applyProbeAvailabilityResultsAt(hostID string, results []Probe Msg("Rejecting probe availability result from an agent that does not own the target") continue } + if result.ConfigRevision > 0 && result.ConfigRevision != target.ConfigRevision { + log.Debug(). + Str("hostID", hostID). + Str("targetID", targetID). + Int64("reportedRevision", result.ConfigRevision). + Int64("currentRevision", target.ConfigRevision). + Msg("Rejecting probe availability result for an obsolete configuration revision") + continue + } checkedAt := result.CheckedAt if checkedAt.IsZero() { @@ -138,7 +153,11 @@ func (m *Monitor) applyProbeAvailabilityResultsAt(hostID string, results []Probe latency = 0 } outcome, probeErr := probeResultOutcome(result) - m.applyAvailabilityObservation(target, checkedAt.UTC(), latency, outcome, probeErr, result.Certificate, hostID, receivedAt) + observationID := strings.TrimSpace(result.ObservationID) + if observationID == "" { + observationID = legacyProbeAvailabilityObservationID(hostID, result) + } + m.applyAvailabilityObservation(target, observationID, checkedAt.UTC(), latency, outcome, probeErr, result.Certificate, hostID, receivedAt) applied++ } @@ -148,6 +167,14 @@ func (m *Monitor) applyProbeAvailabilityResultsAt(hostID string, results []Probe m.updateResourceStore(m.GetState()) } +func legacyProbeAvailabilityObservationID(hostID string, result ProbeAvailabilityResult) string { + material := fmt.Sprintf("%s\x00%s\x00%d\x00%s\x00%d\x00%s", + strings.TrimSpace(hostID), strings.TrimSpace(result.TargetID), result.CheckedAt.UTC().UnixNano(), + strings.TrimSpace(string(result.Outcome)), result.LatencyMillis, strings.TrimSpace(result.Error)) + sum := sha256.Sum256([]byte(material)) + return fmt.Sprintf("legacy-agent-%x", sum[:]) +} + // probeResultOutcome normalizes a reported outcome and derives the failure // signal. An unreachable report without a message still has to fail, otherwise // remote checks would never accumulate consecutive failures. @@ -273,6 +300,7 @@ func (m *Monitor) availabilityProbeTargetsForAgent(hostID string) []map[string]i func availabilityProbeAgentTargetPayload(target config.AvailabilityTarget) map[string]interface{} { payload := map[string]interface{}{ "id": target.ID, + "configRevision": target.ConfigRevision, "name": target.DisplayName(), "targetKind": string(target.TargetKind), "address": target.Address, diff --git a/internal/monitoring/availability_probe_agent_test.go b/internal/monitoring/availability_probe_agent_test.go index 7c277abe1..2d58364d5 100644 --- a/internal/monitoring/availability_probe_agent_test.go +++ b/internal/monitoring/availability_probe_agent_test.go @@ -12,6 +12,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" pkglicensing "github.com/rcourtman/pulse-go-rewrite/pkg/licensing" + pkgmetrics "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" "github.com/rcourtman/pulse-go-rewrite/pkg/tlsutil" ) @@ -233,6 +234,7 @@ func TestGetHostAgentConfigPayloadCarriesProbeParameters(t *testing.T) { } want := map[string]interface{}{ "id": "udp-check", + "configRevision": int64(1), "name": "UDP check", "targetKind": string(config.AvailabilityTargetDevice), "address": "sensor.local", @@ -352,6 +354,56 @@ func TestApplyProbeAvailabilityResultsAccountsFailuresAndAttribution(t *testing. } } +func TestAvailabilityHistoryUsesServerReceiptForRemoteResultsAndDeduplicatesRetries(t *testing.T) { + remoteTarget := probeAgentTarget("remote", "agent-1") + remoteTarget.ConfigRevision = 3 + monitor := newProbeAgentTestMonitor(t, remoteTarget) + monitor.SetLicenseChecker(licenseWithExternalProbe(true)) + store, err := pkgmetrics.NewStore(pkgmetrics.DefaultConfig(t.TempDir())) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + monitor.metricsStore = store + + receivedAt := time.Now().UTC().Truncate(time.Minute).Add(-10 * time.Minute) + result := ProbeAvailabilityResult{ + ObservationID: "agent-observation-1", TargetID: "remote", ConfigRevision: 3, + Outcome: AvailabilityProbeUnreachable, CheckedAt: receivedAt.Add(-24 * time.Hour), + } + monitor.applyProbeAvailabilityResultsAt("agent-1", []ProbeAvailabilityResult{result}, receivedAt) + monitor.applyProbeAvailabilityResultsAt("agent-1", []ProbeAvailabilityResult{result}, receivedAt.Add(time.Second)) + + rows, _, _, err := store.AvailabilityHistoryRowCounts("remote") + if err != nil { + t.Fatal(err) + } + if rows != 1 { + t.Fatalf("remote observation rows = %d, want one idempotent row", rows) + } + history, err := store.QueryAvailabilityHistory( + []string{"remote"}, receivedAt.Add(-time.Minute), receivedAt.Add(6*time.Minute), 120, + ) + if err != nil { + t.Fatal(err) + } + if got := history["remote"].Summary.UnreachableSeconds; got != 300 { + t.Fatalf("remote unreachable duration = %v, want 300 seconds on the server receipt timeline", got) + } + + stale := result + stale.ObservationID = "obsolete-revision" + stale.ConfigRevision = 2 + monitor.applyProbeAvailabilityResultsAt("agent-1", []ProbeAvailabilityResult{stale}, receivedAt.Add(2*time.Second)) + rows, _, _, err = store.AvailabilityHistoryRowCounts("remote") + if err != nil { + t.Fatal(err) + } + if rows != 1 { + t.Fatalf("obsolete revision created a history row; rows = %d", rows) + } +} + func TestApplyProbeAvailabilityResultsIndeterminateClearsFailures(t *testing.T) { monitor := newProbeAgentTestMonitor(t, probeAgentTarget("remote", "agent-1")) monitor.SetLicenseChecker(licenseWithExternalProbe(true)) diff --git a/pkg/agents/host/report.go b/pkg/agents/host/report.go index 61dc39008..bc76d44b1 100644 --- a/pkg/agents/host/report.go +++ b/pkg/agents/host/report.go @@ -136,12 +136,14 @@ type ProxmoxLXCContainer struct { // ("reachable", "unreachable", "indeterminate"); anything else is treated as // indeterminate by the server. type AvailabilityProbeResult struct { - TargetID string `json:"targetId"` - Outcome string `json:"outcome"` - LatencyMillis int64 `json:"latencyMillis"` - CheckedAt time.Time `json:"checkedAt"` - Error string `json:"error,omitempty"` - Certificate *tlsutil.CertificateObservation `json:"certificate,omitempty"` + ObservationID string `json:"observationId,omitempty"` + TargetID string `json:"targetId"` + ConfigRevision int64 `json:"configRevision,omitempty"` + Outcome string `json:"outcome"` + LatencyMillis int64 `json:"latencyMillis"` + CheckedAt time.Time `json:"checkedAt"` + Error string `json:"error,omitempty"` + Certificate *tlsutil.CertificateObservation `json:"certificate,omitempty"` } // ClusterNodeSensors contains temperature sensor data collected from a Proxmox diff --git a/pkg/agents/host/report_test.go b/pkg/agents/host/report_test.go index ce6815b60..d15644571 100644 --- a/pkg/agents/host/report_test.go +++ b/pkg/agents/host/report_test.go @@ -634,7 +634,8 @@ func TestReportAvailabilityResultsJSONRoundTrip(t *testing.T) { Host: HostInfo{Hostname: "probe-agent.local"}, AvailabilityResults: []AvailabilityProbeResult{ { - TargetID: "target-1", Outcome: "reachable", LatencyMillis: 12, CheckedAt: checkedAt, + ObservationID: "observation-1", TargetID: "target-1", ConfigRevision: 7, + Outcome: "reachable", LatencyMillis: 12, CheckedAt: checkedAt, Certificate: &tlsutil.CertificateObservation{ Subject: "pulse.example.test", TrustStatus: tlsutil.CertificateTrustTrusted, ObservedAt: checkedAt, DNSNames: []string{"pulse.example.test"}, @@ -664,7 +665,8 @@ func TestReportAvailabilityResultsJSONRoundTrip(t *testing.T) { t.Fatalf("availability results = %+v, want 2", decoded.AvailabilityResults) } first := decoded.AvailabilityResults[0] - if first.TargetID != "target-1" || first.Outcome != "reachable" || first.LatencyMillis != 12 { + if first.ObservationID != "observation-1" || first.TargetID != "target-1" || + first.ConfigRevision != 7 || first.Outcome != "reachable" || first.LatencyMillis != 12 { t.Fatalf("first result = %+v", first) } if !first.CheckedAt.Equal(checkedAt) { diff --git a/pkg/metrics/availability_history.go b/pkg/metrics/availability_history.go new file mode 100644 index 000000000..84c922531 --- /dev/null +++ b/pkg/metrics/availability_history.go @@ -0,0 +1,843 @@ +package metrics + +import ( + "database/sql" + "fmt" + "math" + "sort" + "strings" + "time" + + pdb "github.com/rcourtman/pulse-go-rewrite/pkg/db" + "github.com/rs/zerolog/log" +) + +// AvailabilityOutcome is the closed categorical vocabulary retained by the +// history store. It deliberately does not reuse numeric metric rows: averaging +// these values would erase indeterminate and unknown time. +type AvailabilityOutcome string + +const ( + AvailabilityReachable AvailabilityOutcome = "reachable" + AvailabilityUnreachable AvailabilityOutcome = "unreachable" + AvailabilityIndeterminate AvailabilityOutcome = "indeterminate" +) + +// AvailabilityExecutionSource identifies where an observation executed +// without retaining the assigned agent's identity. +type AvailabilityExecutionSource string + +const ( + AvailabilitySourceLocal AvailabilityExecutionSource = "local" + AvailabilitySourceAssignedAgent AvailabilityExecutionSource = "assigned_agent" +) + +const ( + availabilityRawRetention = 48 * time.Hour + availabilityMinuteRetention = 8 * 24 * time.Hour + availabilityHourlyRetention = 92 * 24 * time.Hour + availabilityDailyRetention = 366 * 24 * time.Hour + availabilityMaxValidity = 24 * time.Hour + availabilityMaxBatchTargets = 200 + availabilityMaxBuckets = 120 +) + +// AvailabilityObservation is one accepted scheduled result. TimelineAt is +// server-authored and controls coverage; ObservedAt is evidence metadata only. +type AvailabilityObservation struct { + ObservationID string + TargetID string + ConfigRevision int64 + Outcome AvailabilityOutcome + ObservedAt time.Time + TimelineAt time.Time + IngestedAt time.Time + ValidFor time.Duration + ExecutionSource AvailabilityExecutionSource + LatencyMillis *int64 +} + +// AvailabilityLatencySummary contains reachable-only latency evidence. +type AvailabilityLatencySummary struct { + Average float64 `json:"average"` + Min int64 `json:"min"` + Max int64 `json:"max"` +} + +// AvailabilityHistorySummary is derived from state durations over the exact +// requested window. AvailabilityPercent is absent without determinate time. +type AvailabilityHistorySummary struct { + ReachableSeconds float64 `json:"reachableSeconds"` + UnreachableSeconds float64 `json:"unreachableSeconds"` + IndeterminateSeconds float64 `json:"indeterminateSeconds"` + UnknownSeconds float64 `json:"unknownSeconds"` + CoveragePercent float64 `json:"coveragePercent"` + AvailabilityPercent *float64 `json:"availabilityPercent,omitempty"` + ReachableLatencyMillis *AvailabilityLatencySummary `json:"reachableLatencyMillis,omitempty"` +} + +// AvailabilityHistoryBucket is one chronological fleet-view bucket. +type AvailabilityHistoryBucket struct { + Start time.Time `json:"start"` + End time.Time `json:"end"` + ReachableSeconds float64 `json:"reachableSeconds"` + UnreachableSeconds float64 `json:"unreachableSeconds"` + IndeterminateSeconds float64 `json:"indeterminateSeconds"` + UnknownSeconds float64 `json:"unknownSeconds"` + LatencyMillis *AvailabilityLatencySummary `json:"latencyMillis,omitempty"` +} + +// AvailabilityRevisionBoundary prevents a chart from implying an unchanged +// check across execution-defining configuration edits. +type AvailabilityRevisionBoundary struct { + Revision int64 `json:"revision"` + At time.Time `json:"at"` +} + +// AvailabilityHistoryTarget is the source-owned result for one target. +type AvailabilityHistoryTarget struct { + TargetID string `json:"targetId"` + Summary AvailabilityHistorySummary `json:"summary"` + Buckets []AvailabilityHistoryBucket `json:"buckets"` + RevisionBoundaries []AvailabilityRevisionBoundary `json:"revisionBoundaries"` +} + +type availabilityAggregate struct { + reachableMillis int64 + unreachableMillis int64 + indeterminateMillis int64 + latencyCount int64 + latencySum int64 + latencyMin int64 + latencyMax int64 +} + +func (s *Store) initAvailabilityHistorySchema() error { + _, err := s.db.Exec(` + CREATE TABLE IF NOT EXISTS availability_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + observation_id TEXT NOT NULL UNIQUE, + target_id TEXT NOT NULL, + config_revision INTEGER NOT NULL, + outcome TEXT NOT NULL CHECK(outcome IN ('reachable', 'unreachable', 'indeterminate')), + observed_at_ns INTEGER NOT NULL, + timeline_at_ns INTEGER NOT NULL, + ingested_at_ns INTEGER NOT NULL, + valid_until_ns INTEGER NOT NULL, + execution_source TEXT NOT NULL CHECK(execution_source IN ('local', 'assigned_agent')), + latency_millis INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_availability_observations_target_timeline + ON availability_observations(target_id, timeline_at_ns, id); + CREATE INDEX IF NOT EXISTS idx_availability_observations_retention + ON availability_observations(timeline_at_ns); + + CREATE TABLE IF NOT EXISTS availability_history_buckets ( + target_id TEXT NOT NULL, + tier TEXT NOT NULL CHECK(tier IN ('minute', 'hourly', 'daily')), + bucket_start INTEGER NOT NULL, + reachable_millis INTEGER NOT NULL DEFAULT 0, + unreachable_millis INTEGER NOT NULL DEFAULT 0, + indeterminate_millis INTEGER NOT NULL DEFAULT 0, + reachable_count INTEGER NOT NULL DEFAULT 0, + unreachable_count INTEGER NOT NULL DEFAULT 0, + indeterminate_count INTEGER NOT NULL DEFAULT 0, + latency_count INTEGER NOT NULL DEFAULT 0, + latency_sum INTEGER NOT NULL DEFAULT 0, + latency_min INTEGER, + latency_max INTEGER, + PRIMARY KEY(target_id, tier, bucket_start) + ); + CREATE INDEX IF NOT EXISTS idx_availability_buckets_retention + ON availability_history_buckets(tier, bucket_start); + + CREATE TABLE IF NOT EXISTS availability_revision_boundaries ( + target_id TEXT NOT NULL, + revision INTEGER NOT NULL, + started_at_ns INTEGER NOT NULL, + PRIMARY KEY(target_id, revision) + ); + CREATE INDEX IF NOT EXISTS idx_availability_boundaries_target_time + ON availability_revision_boundaries(target_id, started_at_ns); + `) + if err != nil { + return fmt.Errorf("create availability history schema: %w", err) + } + return nil +} + +func normalizeAvailabilityObservation(observation AvailabilityObservation) (AvailabilityObservation, error) { + observation.ObservationID = strings.TrimSpace(observation.ObservationID) + observation.TargetID = strings.TrimSpace(observation.TargetID) + if observation.ObservationID == "" || observation.TargetID == "" { + return AvailabilityObservation{}, fmt.Errorf("availability observation and target ids are required") + } + if observation.ConfigRevision <= 0 { + observation.ConfigRevision = 1 + } + switch observation.Outcome { + case AvailabilityReachable, AvailabilityUnreachable, AvailabilityIndeterminate: + default: + return AvailabilityObservation{}, fmt.Errorf("unsupported availability outcome %q", observation.Outcome) + } + switch observation.ExecutionSource { + case AvailabilitySourceLocal, AvailabilitySourceAssignedAgent: + default: + return AvailabilityObservation{}, fmt.Errorf("unsupported availability execution source %q", observation.ExecutionSource) + } + if observation.TimelineAt.IsZero() || observation.IngestedAt.IsZero() { + return AvailabilityObservation{}, fmt.Errorf("availability timeline and ingestion times are required") + } + observation.TimelineAt = observation.TimelineAt.UTC() + observation.IngestedAt = observation.IngestedAt.UTC() + if observation.ObservedAt.IsZero() { + observation.ObservedAt = observation.TimelineAt + } else { + observation.ObservedAt = observation.ObservedAt.UTC() + } + if observation.ValidFor <= 0 { + return AvailabilityObservation{}, fmt.Errorf("availability validity window must be positive") + } + if observation.ValidFor > availabilityMaxValidity { + observation.ValidFor = availabilityMaxValidity + } + if observation.LatencyMillis != nil { + latency := *observation.LatencyMillis + if observation.Outcome != AvailabilityReachable || latency < 0 { + observation.LatencyMillis = nil + } else if latency > int64(availabilityMaxValidity/time.Millisecond) { + latency = int64(availabilityMaxValidity / time.Millisecond) + observation.LatencyMillis = &latency + } + } + return observation, nil +} + +// WriteAvailabilityObservationSync is the read-your-writes test and seeding +// path. Live monitoring uses the bounded variant below. +func (s *Store) WriteAvailabilityObservationSync(observation AvailabilityObservation) error { + normalized, err := normalizeAvailabilityObservation(observation) + if err != nil { + return err + } + s.enqueueAndWait(writeRequest{availability: []AvailabilityObservation{normalized}}) + return nil +} + +// WriteAvailabilityObservationBounded keeps a slow history disk from blocking +// polling. Once queued, the observation remains ordered with numeric writes. +func (s *Store) WriteAvailabilityObservationBounded(observation AvailabilityObservation) error { + normalized, err := normalizeAvailabilityObservation(observation) + if err != nil { + return err + } + s.boundedEnqueueAndWait(writeRequest{availability: []AvailabilityObservation{normalized}}) + return nil +} + +// DeleteAvailabilityTargetHistory removes raw observations, compact buckets, +// and revision boundaries behind the same writer barrier as ingestion. +func (s *Store) DeleteAvailabilityTargetHistory(targetID string) { + targetID = strings.TrimSpace(targetID) + if s == nil || targetID == "" || s.stopping.Load() { + return + } + s.enqueueAndWait(writeRequest{availabilityDeletes: []string{targetID}}) +} + +func (s *Store) writeAvailabilityBatch(observations []AvailabilityObservation) { + if len(observations) == 0 { + return + } + tx, err := s.db.Begin() + if err != nil { + log.Error().Err(err).Msg("Failed to begin availability history transaction") + return + } + defer func() { _ = tx.Rollback() }() + + seen := make(map[string]struct{}, len(observations)) + for _, observation := range observations { + normalized, err := normalizeAvailabilityObservation(observation) + if err != nil { + log.Warn().Err(err).Msg("Dropping invalid availability observation") + continue + } + if _, exists := seen[normalized.ObservationID]; exists { + continue + } + seen[normalized.ObservationID] = struct{}{} + if err := insertAvailabilityObservation(tx, normalized); err != nil { + log.Warn().Err(err).Str("target_id", normalized.TargetID).Msg("Failed to write availability observation") + } + } + if err := tx.Commit(); err != nil { + log.Error().Err(err).Int("batch_size", len(observations)).Msg("Failed to commit availability history batch") + } +} + +func insertAvailabilityObservation(tx *pdb.InstrumentedTx, observation AvailabilityObservation) error { + var latency any + if observation.LatencyMillis != nil { + latency = *observation.LatencyMillis + } + result, err := tx.Exec(` + INSERT OR IGNORE INTO availability_observations ( + observation_id, target_id, config_revision, outcome, observed_at_ns, + timeline_at_ns, ingested_at_ns, valid_until_ns, execution_source, latency_millis + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, observation.ObservationID, observation.TargetID, observation.ConfigRevision, string(observation.Outcome), + observation.ObservedAt.UnixNano(), observation.TimelineAt.UnixNano(), observation.IngestedAt.UnixNano(), + observation.TimelineAt.Add(observation.ValidFor).UnixNano(), string(observation.ExecutionSource), latency) + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil || affected == 0 { + return err + } + rowID, err := result.LastInsertId() + if err != nil { + return err + } + + if _, err := tx.Exec(` + INSERT OR IGNORE INTO availability_revision_boundaries (target_id, revision, started_at_ns) + VALUES (?, ?, ?) + `, observation.TargetID, observation.ConfigRevision, observation.TimelineAt.UnixNano()); err != nil { + return err + } + if err := addAvailabilityObservationAggregates(tx, observation); err != nil { + return err + } + + var previousOutcome string + var previousStart, previousValidUntil int64 + err = tx.QueryRow(` + SELECT outcome, timeline_at_ns, valid_until_ns + FROM availability_observations + WHERE target_id = ? AND id < ? + ORDER BY id DESC LIMIT 1 + `, observation.TargetID, rowID).Scan(&previousOutcome, &previousStart, &previousValidUntil) + if err == sql.ErrNoRows { + return nil + } + if err != nil { + return err + } + end := observation.TimelineAt.UnixNano() + if previousValidUntil < end { + end = previousValidUntil + } + if end <= previousStart { + return nil + } + return addAvailabilityDurationAggregates(tx, observation.TargetID, AvailabilityOutcome(previousOutcome), previousStart, end) +} + +var availabilityStorageTiers = []struct { + tier string + size time.Duration +}{ + {string(TierMinute), time.Minute}, + {string(TierHourly), time.Hour}, + {string(TierDaily), 24 * time.Hour}, +} + +func addAvailabilityObservationAggregates(tx *pdb.InstrumentedTx, observation AvailabilityObservation) error { + for _, tier := range availabilityStorageTiers { + bucketStart := observation.TimelineAt.Truncate(tier.size).Unix() + reachableCount, unreachableCount, indeterminateCount := int64(0), int64(0), int64(0) + switch observation.Outcome { + case AvailabilityReachable: + reachableCount = 1 + case AvailabilityUnreachable: + unreachableCount = 1 + case AvailabilityIndeterminate: + indeterminateCount = 1 + } + latencyCount, latencySum := int64(0), int64(0) + var latencyMin, latencyMax any + if observation.Outcome == AvailabilityReachable && observation.LatencyMillis != nil { + latencyCount = 1 + latencySum = *observation.LatencyMillis + latencyMin = *observation.LatencyMillis + latencyMax = *observation.LatencyMillis + } + _, err := tx.Exec(` + INSERT INTO availability_history_buckets ( + target_id, tier, bucket_start, reachable_count, unreachable_count, + indeterminate_count, latency_count, latency_sum, latency_min, latency_max + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(target_id, tier, bucket_start) DO UPDATE SET + reachable_count = reachable_count + excluded.reachable_count, + unreachable_count = unreachable_count + excluded.unreachable_count, + indeterminate_count = indeterminate_count + excluded.indeterminate_count, + latency_count = latency_count + excluded.latency_count, + latency_sum = latency_sum + excluded.latency_sum, + latency_min = CASE WHEN excluded.latency_min IS NULL THEN latency_min + WHEN latency_min IS NULL THEN excluded.latency_min ELSE MIN(latency_min, excluded.latency_min) END, + latency_max = CASE WHEN excluded.latency_max IS NULL THEN latency_max + WHEN latency_max IS NULL THEN excluded.latency_max ELSE MAX(latency_max, excluded.latency_max) END + `, observation.TargetID, tier.tier, bucketStart, reachableCount, unreachableCount, + indeterminateCount, latencyCount, latencySum, latencyMin, latencyMax) + if err != nil { + return err + } + } + return nil +} + +func addAvailabilityDurationAggregates(tx *pdb.InstrumentedTx, targetID string, outcome AvailabilityOutcome, startNS, endNS int64) error { + for _, tier := range availabilityStorageTiers { + cursor := time.Unix(0, startNS).UTC() + end := time.Unix(0, endNS).UTC() + for cursor.Before(end) { + bucketStart := cursor.Truncate(tier.size) + segmentEnd := bucketStart.Add(tier.size) + if segmentEnd.After(end) { + segmentEnd = end + } + millis := segmentEnd.Sub(cursor).Milliseconds() + if millis > 0 { + reachable, unreachable, indeterminate := int64(0), int64(0), int64(0) + switch outcome { + case AvailabilityReachable: + reachable = millis + case AvailabilityUnreachable: + unreachable = millis + case AvailabilityIndeterminate: + indeterminate = millis + } + if _, err := tx.Exec(` + INSERT INTO availability_history_buckets ( + target_id, tier, bucket_start, reachable_millis, unreachable_millis, indeterminate_millis + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(target_id, tier, bucket_start) DO UPDATE SET + reachable_millis = reachable_millis + excluded.reachable_millis, + unreachable_millis = unreachable_millis + excluded.unreachable_millis, + indeterminate_millis = indeterminate_millis + excluded.indeterminate_millis + `, targetID, tier.tier, bucketStart.Unix(), reachable, unreachable, indeterminate); err != nil { + return err + } + } + cursor = segmentEnd + } + } + return nil +} + +func (s *Store) deleteAvailabilityTargets(targetIDs []string) { + tx, err := s.db.Begin() + if err != nil { + log.Warn().Err(err).Msg("Failed to begin availability history deletion") + return + } + defer func() { _ = tx.Rollback() }() + for _, targetID := range targetIDs { + targetID = strings.TrimSpace(targetID) + if targetID == "" { + continue + } + for _, statement := range []string{ + `DELETE FROM availability_observations WHERE target_id = ?`, + `DELETE FROM availability_history_buckets WHERE target_id = ?`, + `DELETE FROM availability_revision_boundaries WHERE target_id = ?`, + } { + if _, err := tx.Exec(statement, targetID); err != nil { + log.Warn().Err(err).Str("target_id", targetID).Msg("Failed to delete availability history") + return + } + } + } + if err := tx.Commit(); err != nil { + log.Warn().Err(err).Msg("Failed to commit availability history deletion") + } +} + +func (s *Store) runAvailabilityRetention() { + now := time.Now().UTC() + if _, err := s.db.Exec(`DELETE FROM availability_observations WHERE timeline_at_ns < ?`, + now.Add(-s.effectiveRetention(availabilityRawRetention, now)).UnixNano()); err != nil { + log.Warn().Err(err).Msg("Failed to prune raw availability observations") + } + for _, tier := range []struct { + name string + retention time.Duration + }{ + {string(TierMinute), availabilityMinuteRetention}, + {string(TierHourly), availabilityHourlyRetention}, + {string(TierDaily), availabilityDailyRetention}, + } { + cutoff := now.Add(-s.effectiveRetention(tier.retention, now)).Unix() + if _, err := s.db.Exec(`DELETE FROM availability_history_buckets WHERE tier = ? AND bucket_start < ?`, tier.name, cutoff); err != nil { + log.Warn().Err(err).Str("tier", tier.name).Msg("Failed to prune availability history buckets") + } + } + cutoff := now.Add(-s.effectiveRetention(availabilityDailyRetention, now)).UnixNano() + if _, err := s.db.Exec(`DELETE FROM availability_revision_boundaries WHERE started_at_ns < ?`, cutoff); err != nil { + log.Warn().Err(err).Msg("Failed to prune availability revision boundaries") + } +} + +// QueryAvailabilityHistory reads all requested targets with three bounded +// batch queries (buckets, live tails, and revision boundaries), never one +// query per target. +func (s *Store) QueryAvailabilityHistory(targetIDs []string, start, end time.Time, maxBuckets int) (map[string]AvailabilityHistoryTarget, error) { + ids := normalizeAvailabilityTargetIDs(targetIDs) + if len(ids) == 0 { + return map[string]AvailabilityHistoryTarget{}, nil + } + if len(ids) > availabilityMaxBatchTargets { + return nil, fmt.Errorf("availability history supports at most %d target ids", availabilityMaxBatchTargets) + } + start, end = start.UTC(), end.UTC() + if !end.After(start) { + return nil, fmt.Errorf("availability history end must be after start") + } + if maxBuckets <= 0 || maxBuckets > availabilityMaxBuckets { + maxBuckets = availabilityMaxBuckets + } + + step := availabilityPresentationStep(end.Sub(start), maxBuckets) + results := make(map[string]AvailabilityHistoryTarget, len(ids)) + aggregates := make(map[string][]availabilityAggregate, len(ids)) + for _, id := range ids { + buckets := makeAvailabilityBuckets(start, end, step) + results[id] = AvailabilityHistoryTarget{TargetID: id, Buckets: buckets, RevisionBoundaries: []AvailabilityRevisionBoundary{}} + aggregates[id] = make([]availabilityAggregate, len(buckets)) + } + + if err := s.queryAvailabilityBuckets(ids, start, end, step, aggregates); err != nil { + return nil, err + } + if err := s.queryAvailabilityLiveTails(ids, start, end, step, aggregates); err != nil { + return nil, err + } + boundaries, err := s.queryAvailabilityRevisionBoundaries(ids, start, end) + if err != nil { + return nil, err + } + + windowMillis := end.Sub(start).Milliseconds() + for _, id := range ids { + target := results[id] + total := availabilityAggregate{} + for index := range target.Buckets { + aggregate := aggregates[id][index] + bucketMillis := target.Buckets[index].End.Sub(target.Buckets[index].Start).Milliseconds() + target.Buckets[index] = finalizeAvailabilityBucket(target.Buckets[index], aggregate, bucketMillis) + mergeAvailabilityAggregate(&total, aggregate) + } + target.Summary = finalizeAvailabilitySummary(total, windowMillis) + target.RevisionBoundaries = boundaries[id] + results[id] = target + } + return results, nil +} + +func normalizeAvailabilityTargetIDs(targetIDs []string) []string { + seen := make(map[string]struct{}, len(targetIDs)) + ids := make([]string, 0, len(targetIDs)) + for _, id := range targetIDs { + id = strings.TrimSpace(id) + if id == "" { + continue + } + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + ids = append(ids, id) + } + return ids +} + +func availabilityPresentationStep(window time.Duration, maxBuckets int) time.Duration { + steps := []time.Duration{time.Minute, 2 * time.Minute, 5 * time.Minute, 6 * time.Minute, 12 * time.Minute, 30 * time.Minute, time.Hour, 2 * time.Hour, 6 * time.Hour, 12 * time.Hour, 24 * time.Hour, 4 * 24 * time.Hour} + minimum := time.Duration(math.Ceil(float64(window) / float64(maxBuckets))) + for _, step := range steps { + if step >= minimum { + return step + } + } + return time.Duration(math.Ceil(float64(minimum)/(float64(24*time.Hour)))) * 24 * time.Hour +} + +func makeAvailabilityBuckets(start, end time.Time, step time.Duration) []AvailabilityHistoryBucket { + count := int(math.Ceil(float64(end.Sub(start)) / float64(step))) + buckets := make([]AvailabilityHistoryBucket, 0, count) + for cursor := start; cursor.Before(end); cursor = cursor.Add(step) { + bucketEnd := cursor.Add(step) + if bucketEnd.After(end) { + bucketEnd = end + } + buckets = append(buckets, AvailabilityHistoryBucket{Start: cursor, End: bucketEnd}) + } + return buckets +} + +func availabilityPlaceholders(count int) string { + return strings.TrimSuffix(strings.Repeat("?,", count), ",") +} + +func (s *Store) queryAvailabilityBuckets(ids []string, start, end time.Time, step time.Duration, aggregates map[string][]availabilityAggregate) error { + minuteStart := start + if candidate := end.Add(-7 * 24 * time.Hour); candidate.After(minuteStart) { + minuteStart = candidate.Truncate(time.Hour) + } + hourlyStart := start + if candidate := end.Add(-90 * 24 * time.Hour); candidate.After(hourlyStart) { + hourlyStart = candidate.Truncate(24 * time.Hour) + } + + args := make([]any, 0, len(ids)+9) + for _, id := range ids { + args = append(args, id) + } + args = append(args, + string(TierDaily), start.Unix(), hourlyStart.Unix(), + string(TierHourly), hourlyStart.Unix(), minuteStart.Unix(), + string(TierMinute), minuteStart.Unix(), end.Unix(), + ) + rows, err := s.db.Query(fmt.Sprintf(` + SELECT target_id, bucket_start, reachable_millis, unreachable_millis, + indeterminate_millis, latency_count, latency_sum, + COALESCE(latency_min, 0), COALESCE(latency_max, 0) + FROM availability_history_buckets + WHERE target_id IN (%s) AND ( + (tier = ? AND bucket_start >= ? AND bucket_start < ?) OR + (tier = ? AND bucket_start >= ? AND bucket_start < ?) OR + (tier = ? AND bucket_start >= ? AND bucket_start < ?) + ) + ORDER BY target_id, bucket_start + `, availabilityPlaceholders(len(ids))), args...) + if err != nil { + return fmt.Errorf("query availability history buckets: %w", err) + } + defer rows.Close() + for rows.Next() { + var targetID string + var bucketStart, reachable, unreachable, indeterminate, latencyCount, latencySum, latencyMin, latencyMax int64 + if err := rows.Scan(&targetID, &bucketStart, &reachable, &unreachable, &indeterminate, &latencyCount, &latencySum, &latencyMin, &latencyMax); err != nil { + return err + } + index := int(time.Unix(bucketStart, 0).Sub(start) / step) + if index < 0 || index >= len(aggregates[targetID]) { + continue + } + mergeAvailabilityAggregate(&aggregates[targetID][index], availabilityAggregate{ + reachableMillis: reachable, unreachableMillis: unreachable, indeterminateMillis: indeterminate, + latencyCount: latencyCount, latencySum: latencySum, latencyMin: latencyMin, latencyMax: latencyMax, + }) + } + return rows.Err() +} + +func (s *Store) queryAvailabilityLiveTails(ids []string, start, end time.Time, step time.Duration, aggregates map[string][]availabilityAggregate) error { + args := make([]any, 0, len(ids)) + for _, id := range ids { + args = append(args, id) + } + rows, err := s.db.Query(fmt.Sprintf(` + WITH ranked AS ( + SELECT target_id, outcome, timeline_at_ns, valid_until_ns, + ROW_NUMBER() OVER (PARTITION BY target_id ORDER BY timeline_at_ns DESC, id DESC) AS rank + FROM availability_observations WHERE target_id IN (%s) + ) + SELECT target_id, outcome, timeline_at_ns, valid_until_ns FROM ranked WHERE rank = 1 + `, availabilityPlaceholders(len(ids))), args...) + if err != nil { + return fmt.Errorf("query availability live tails: %w", err) + } + defer rows.Close() + for rows.Next() { + var targetID, outcome string + var tailStartNS, validUntilNS int64 + if err := rows.Scan(&targetID, &outcome, &tailStartNS, &validUntilNS); err != nil { + return err + } + tailStart := time.Unix(0, tailStartNS).UTC() + if tailStart.Before(start) { + tailStart = start + } + tailEnd := time.Unix(0, validUntilNS).UTC() + if tailEnd.After(end) { + tailEnd = end + } + addAvailabilityRangeToPresentation(aggregates[targetID], start, step, AvailabilityOutcome(outcome), tailStart, tailEnd) + } + return rows.Err() +} + +func addAvailabilityRangeToPresentation(buckets []availabilityAggregate, start time.Time, step time.Duration, outcome AvailabilityOutcome, cursor, end time.Time) { + if !end.After(cursor) { + return + } + for cursor.Before(end) { + index := int(cursor.Sub(start) / step) + if index < 0 || index >= len(buckets) { + return + } + segmentEnd := start.Add(time.Duration(index+1) * step) + if segmentEnd.After(end) { + segmentEnd = end + } + millis := segmentEnd.Sub(cursor).Milliseconds() + switch outcome { + case AvailabilityReachable: + buckets[index].reachableMillis += millis + case AvailabilityUnreachable: + buckets[index].unreachableMillis += millis + case AvailabilityIndeterminate: + buckets[index].indeterminateMillis += millis + } + cursor = segmentEnd + } +} + +func (s *Store) queryAvailabilityRevisionBoundaries(ids []string, start, end time.Time) (map[string][]AvailabilityRevisionBoundary, error) { + args := make([]any, 0, len(ids)+2) + for _, id := range ids { + args = append(args, id) + } + args = append(args, start.UnixNano(), end.UnixNano()) + rows, err := s.db.Query(fmt.Sprintf(` + SELECT target_id, revision, started_at_ns + FROM availability_revision_boundaries + WHERE target_id IN (%s) AND started_at_ns >= ? AND started_at_ns < ? + ORDER BY target_id, started_at_ns + `, availabilityPlaceholders(len(ids))), args...) + if err != nil { + return nil, fmt.Errorf("query availability revision boundaries: %w", err) + } + defer rows.Close() + result := make(map[string][]AvailabilityRevisionBoundary, len(ids)) + for rows.Next() { + var targetID string + var revision, at int64 + if err := rows.Scan(&targetID, &revision, &at); err != nil { + return nil, err + } + result[targetID] = append(result[targetID], AvailabilityRevisionBoundary{Revision: revision, At: time.Unix(0, at).UTC()}) + } + return result, rows.Err() +} + +func mergeAvailabilityAggregate(target *availabilityAggregate, source availabilityAggregate) { + target.reachableMillis += source.reachableMillis + target.unreachableMillis += source.unreachableMillis + target.indeterminateMillis += source.indeterminateMillis + if source.latencyCount > 0 { + if target.latencyCount == 0 || source.latencyMin < target.latencyMin { + target.latencyMin = source.latencyMin + } + if target.latencyCount == 0 || source.latencyMax > target.latencyMax { + target.latencyMax = source.latencyMax + } + target.latencyCount += source.latencyCount + target.latencySum += source.latencySum + } +} + +func finalizeAvailabilityBucket(bucket AvailabilityHistoryBucket, aggregate availabilityAggregate, bucketMillis int64) AvailabilityHistoryBucket { + known := aggregate.reachableMillis + aggregate.unreachableMillis + aggregate.indeterminateMillis + if known > bucketMillis { + known = bucketMillis + } + bucket.ReachableSeconds = millisecondsToSeconds(aggregate.reachableMillis) + bucket.UnreachableSeconds = millisecondsToSeconds(aggregate.unreachableMillis) + bucket.IndeterminateSeconds = millisecondsToSeconds(aggregate.indeterminateMillis) + bucket.UnknownSeconds = millisecondsToSeconds(bucketMillis - known) + bucket.LatencyMillis = availabilityLatencySummary(aggregate) + return bucket +} + +func finalizeAvailabilitySummary(aggregate availabilityAggregate, windowMillis int64) AvailabilityHistorySummary { + known := aggregate.reachableMillis + aggregate.unreachableMillis + aggregate.indeterminateMillis + if known > windowMillis { + known = windowMillis + } + summary := AvailabilityHistorySummary{ + ReachableSeconds: millisecondsToSeconds(aggregate.reachableMillis), + UnreachableSeconds: millisecondsToSeconds(aggregate.unreachableMillis), + IndeterminateSeconds: millisecondsToSeconds(aggregate.indeterminateMillis), + UnknownSeconds: millisecondsToSeconds(windowMillis - known), + ReachableLatencyMillis: availabilityLatencySummary(aggregate), + } + if windowMillis > 0 { + summary.CoveragePercent = roundPercent(float64(known) * 100 / float64(windowMillis)) + } + determinate := aggregate.reachableMillis + aggregate.unreachableMillis + if determinate > 0 { + value := roundPercent(float64(aggregate.reachableMillis) * 100 / float64(determinate)) + summary.AvailabilityPercent = &value + } + return summary +} + +func availabilityLatencySummary(aggregate availabilityAggregate) *AvailabilityLatencySummary { + if aggregate.latencyCount <= 0 { + return nil + } + return &AvailabilityLatencySummary{ + Average: math.Round(float64(aggregate.latencySum)/float64(aggregate.latencyCount)*100) / 100, + Min: aggregate.latencyMin, + Max: aggregate.latencyMax, + } +} + +func millisecondsToSeconds(value int64) float64 { + if value <= 0 { + return 0 + } + return math.Round(float64(value)/10) / 100 +} + +func roundPercent(value float64) float64 { + return math.Round(value*100) / 100 +} + +// AvailabilityHistoryRowCounts is a narrow diagnostic used by contract tests +// to prove idempotency and full target deletion without exposing stored data. +func (s *Store) AvailabilityHistoryRowCounts(targetID string) (raw, buckets, boundaries int64, err error) { + targetID = strings.TrimSpace(targetID) + queries := []struct { + statement string + value *int64 + }{ + {`SELECT COUNT(*) FROM availability_observations WHERE target_id = ?`, &raw}, + {`SELECT COUNT(*) FROM availability_history_buckets WHERE target_id = ?`, &buckets}, + {`SELECT COUNT(*) FROM availability_revision_boundaries WHERE target_id = ?`, &boundaries}, + } + for _, query := range queries { + if scanErr := s.db.QueryRow(query.statement, targetID).Scan(query.value); scanErr != nil { + return 0, 0, 0, scanErr + } + } + return raw, buckets, boundaries, nil +} + +// SortedAvailabilityHistoryTargets converts the batch map to stable input +// order when callers need a slice response. +func SortedAvailabilityHistoryTargets(results map[string]AvailabilityHistoryTarget, order []string) []AvailabilityHistoryTarget { + items := make([]AvailabilityHistoryTarget, 0, len(results)) + seen := make(map[string]struct{}, len(results)) + for _, id := range order { + if item, ok := results[id]; ok { + items = append(items, item) + seen[id] = struct{}{} + } + } + remaining := make([]string, 0, len(results)-len(seen)) + for id := range results { + if _, ok := seen[id]; !ok { + remaining = append(remaining, id) + } + } + sort.Strings(remaining) + for _, id := range remaining { + items = append(items, results[id]) + } + return items +} diff --git a/pkg/metrics/availability_history_test.go b/pkg/metrics/availability_history_test.go new file mode 100644 index 000000000..d0877ee5c --- /dev/null +++ b/pkg/metrics/availability_history_test.go @@ -0,0 +1,212 @@ +package metrics + +import ( + "fmt" + "math" + "strings" + "testing" + "time" +) + +func TestAvailabilityHistorySchemaDoesNotRetainTargetOrAgentDetails(t *testing.T) { + store, _ := newAvailabilityHistoryTestStore(t) + rows, err := store.db.Query(`PRAGMA table_info(availability_observations)`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + for rows.Next() { + var cid, notNull, primaryKey int + var name, dataType string + var defaultValue any + if err := rows.Scan(&cid, &name, &dataType, ¬Null, &defaultValue, &primaryKey); err != nil { + t.Fatal(err) + } + lower := strings.ToLower(name) + for _, forbidden := range []string{"address", "error", "agent_id", "agent_name"} { + if strings.Contains(lower, forbidden) { + t.Fatalf("availability observation schema retains forbidden detail in column %q", name) + } + } + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } +} + +func newAvailabilityHistoryTestStore(t *testing.T) (*Store, StoreConfig) { + t.Helper() + config := DefaultConfig(t.TempDir()) + store, err := NewStore(config) + if err != nil { + t.Fatalf("NewStore() error = %v", err) + } + t.Cleanup(func() { _ = store.Close() }) + return store, config +} + +func availabilityObservation(id, target string, revision int64, outcome AvailabilityOutcome, at time.Time, validFor time.Duration, latency *int64) AvailabilityObservation { + return AvailabilityObservation{ + ObservationID: id, + TargetID: target, + ConfigRevision: revision, + Outcome: outcome, + ObservedAt: at, + TimelineAt: at, + IngestedAt: at, + ValidFor: validFor, + ExecutionSource: AvailabilitySourceLocal, + LatencyMillis: latency, + } +} + +func TestAvailabilityHistoryCoverageAndLatencyRemainCategorical(t *testing.T) { + store, _ := newAvailabilityHistoryTestStore(t) + end := time.Now().UTC().Truncate(time.Minute) + start := end.Add(-time.Hour) + latency := int64(18) + if err := store.WriteAvailabilityObservationSync(availabilityObservation("one", "target", 1, AvailabilityReachable, end.Add(-30*time.Minute), 5*time.Minute, &latency)); err != nil { + t.Fatal(err) + } + if err := store.WriteAvailabilityObservationSync(availabilityObservation("two", "target", 1, AvailabilityUnreachable, end.Add(-20*time.Minute), 5*time.Minute, nil)); err != nil { + t.Fatal(err) + } + + result, err := store.QueryAvailabilityHistory([]string{"target"}, start, end, 120) + if err != nil { + t.Fatal(err) + } + summary := result["target"].Summary + if summary.ReachableSeconds != 300 || summary.UnreachableSeconds != 300 || summary.IndeterminateSeconds != 0 || summary.UnknownSeconds != 3000 { + t.Fatalf("summary durations = %+v", summary) + } + if summary.AvailabilityPercent == nil || *summary.AvailabilityPercent != 50 { + t.Fatalf("availability = %v, want 50", summary.AvailabilityPercent) + } + if math.Abs(summary.CoveragePercent-16.67) > 0.001 { + t.Fatalf("coverage = %v, want 16.67", summary.CoveragePercent) + } + if summary.ReachableLatencyMillis == nil || summary.ReachableLatencyMillis.Average != 18 { + t.Fatalf("reachable latency = %+v", summary.ReachableLatencyMillis) + } + if len(result["target"].Buckets) > 120 { + t.Fatalf("bucket count = %d, want <= 120", len(result["target"].Buckets)) + } +} + +func TestAvailabilityHistoryIndeterminateAndUnknownNeverCountAsAvailability(t *testing.T) { + store, _ := newAvailabilityHistoryTestStore(t) + end := time.Now().UTC().Truncate(time.Minute) + start := end.Add(-time.Hour) + if err := store.WriteAvailabilityObservationSync(availabilityObservation("udp", "target", 1, AvailabilityIndeterminate, end.Add(-10*time.Minute), 5*time.Minute, nil)); err != nil { + t.Fatal(err) + } + result, err := store.QueryAvailabilityHistory([]string{"target"}, start, end, 120) + if err != nil { + t.Fatal(err) + } + summary := result["target"].Summary + if summary.IndeterminateSeconds != 300 || summary.UnknownSeconds != 3300 { + t.Fatalf("summary = %+v", summary) + } + if summary.AvailabilityPercent != nil { + t.Fatalf("availability = %v, want absent without determinate time", *summary.AvailabilityPercent) + } +} + +func TestAvailabilityHistoryDuplicateRestartRevisionAndDelete(t *testing.T) { + dir := t.TempDir() + config := DefaultConfig(dir) + store, err := NewStore(config) + if err != nil { + t.Fatal(err) + } + base := time.Now().UTC().Truncate(time.Minute).Add(-20 * time.Minute) + first := availabilityObservation("stable-id", "target", 1, AvailabilityReachable, base, 5*time.Minute, nil) + if err := store.WriteAvailabilityObservationSync(first); err != nil { + t.Fatal(err) + } + if err := store.WriteAvailabilityObservationSync(first); err != nil { + t.Fatal(err) + } + if err := store.WriteAvailabilityObservationSync(availabilityObservation("revision-two", "target", 2, AvailabilityUnreachable, base.Add(10*time.Minute), 5*time.Minute, nil)); err != nil { + t.Fatal(err) + } + raw, _, boundaries, err := store.AvailabilityHistoryRowCounts("target") + if err != nil { + t.Fatal(err) + } + if raw != 2 || boundaries != 2 { + t.Fatalf("rows raw=%d boundaries=%d, want 2/2", raw, boundaries) + } + if err := store.Close(); err != nil { + t.Fatal(err) + } + + restarted, err := NewStore(config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = restarted.Close() }) + result, err := restarted.QueryAvailabilityHistory([]string{"target"}, base.Add(-time.Minute), base.Add(20*time.Minute), 120) + if err != nil { + t.Fatal(err) + } + if len(result["target"].RevisionBoundaries) != 2 { + t.Fatalf("revision boundaries = %+v", result["target"].RevisionBoundaries) + } + restarted.DeleteAvailabilityTargetHistory("target") + raw, buckets, boundaries, err := restarted.AvailabilityHistoryRowCounts("target") + if err != nil { + t.Fatal(err) + } + if raw != 0 || buckets != 0 || boundaries != 0 { + t.Fatalf("rows after delete raw=%d buckets=%d boundaries=%d", raw, buckets, boundaries) + } +} + +func TestAvailabilityHistoryBatchReadIsBoundedAtTwoHundredTargets(t *testing.T) { + store, _ := newAvailabilityHistoryTestStore(t) + ids := make([]string, availabilityMaxBatchTargets) + for index := range ids { + ids[index] = fmt.Sprintf("target-%03d", index) + } + end := time.Now().UTC().Truncate(time.Minute) + results, err := store.QueryAvailabilityHistory(ids, end.Add(-24*time.Hour), end, availabilityMaxBuckets) + if err != nil { + t.Fatal(err) + } + if len(results) != availabilityMaxBatchTargets { + t.Fatalf("results = %d, want %d", len(results), availabilityMaxBatchTargets) + } + for _, id := range ids { + if len(results[id].Buckets) > availabilityMaxBuckets { + t.Fatalf("%s buckets = %d", id, len(results[id].Buckets)) + } + if results[id].Summary.CoveragePercent != 0 || results[id].Summary.UnknownSeconds != 86400 { + t.Fatalf("%s unknown summary = %+v", id, results[id].Summary) + } + } + if _, err := store.QueryAvailabilityHistory(append(ids, "target-over-limit"), end.Add(-time.Hour), end, availabilityMaxBuckets); err == nil { + t.Fatal("expected over-limit target batch to fail") + } +} + +func TestAvailabilityHistoryRetentionRemovesRawBucketsAndBoundaries(t *testing.T) { + store, _ := newAvailabilityHistoryTestStore(t) + old := time.Now().UTC().Add(-400 * 24 * time.Hour) + if err := store.WriteAvailabilityObservationSync(availabilityObservation("old-one", "old-target", 1, AvailabilityReachable, old, 5*time.Minute, nil)); err != nil { + t.Fatal(err) + } + if err := store.WriteAvailabilityObservationSync(availabilityObservation("old-two", "old-target", 2, AvailabilityUnreachable, old.Add(10*time.Minute), 5*time.Minute, nil)); err != nil { + t.Fatal(err) + } + store.runAvailabilityRetention() + raw, buckets, boundaries, err := store.AvailabilityHistoryRowCounts("old-target") + if err != nil { + t.Fatal(err) + } + if raw != 0 || buckets != 0 || boundaries != 0 { + t.Fatalf("retained old rows raw=%d buckets=%d boundaries=%d", raw, buckets, boundaries) + } +} diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index 3265d12b2..dde7b9d83 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -138,8 +138,10 @@ type bufferedMetric struct { } type writeRequest struct { - metrics []bufferedMetric - done chan struct{} + metrics []bufferedMetric + availability []AvailabilityObservation + availabilityDeletes []string + done chan struct{} } type metricBatchKey struct { @@ -371,6 +373,9 @@ func (s *Store) initSchema() error { if err := s.ensureMetricsIdentityIndex(); err != nil { return err } + if err := s.initAvailabilityHistorySchema(); err != nil { + return err + } log.Debug().Msg("Metrics schema initialized") return nil @@ -892,6 +897,7 @@ func (s *Store) runStartupMaintenance() { // cleanup trims stale rows and redundant-index pages before SQLite // potentially rewrites the file. s.runRetention() + s.runAvailabilityRetention() s.migrateAutoVacuum() log.Info().Dur("duration", time.Since(start)).Msg("Deferred metrics startup maintenance completed") @@ -918,7 +924,7 @@ func (s *Store) flushLocked() { } func (s *Store) enqueueWrite(req writeRequest) { - if len(req.metrics) == 0 && req.done == nil { + if len(req.metrics) == 0 && len(req.availability) == 0 && len(req.availabilityDeletes) == 0 && req.done == nil { return } @@ -928,7 +934,7 @@ func (s *Store) enqueueWrite(req writeRequest) { log.Warn(). Str("component", "metrics_store"). Str("action", "drop_write_batch"). - Int("batch_size", len(req.metrics)). + Int("batch_size", len(req.metrics)+len(req.availability)+len(req.availabilityDeletes)). Int("write_queue_depth", len(s.writeCh)). Int("write_queue_capacity", cap(s.writeCh)). Msg("Metrics write channel full, dropping batch") @@ -989,7 +995,7 @@ func (s *Store) boundedEnqueueAndWait(req writeRequest) { log.Warn(). Str("component", "metrics_store"). Str("action", "drop_sync_write_batch"). - Int("batch_size", len(req.metrics)). + Int("batch_size", len(req.metrics)+len(req.availability)+len(req.availabilityDeletes)). Int("write_queue_depth", len(s.writeCh)). Int("write_queue_capacity", cap(s.writeCh)). Msg("Metrics write queue saturated, dropping bounded batch to keep monitoring live") @@ -1000,7 +1006,7 @@ func (s *Store) boundedEnqueueAndWait(req writeRequest) { case <-req.done: case <-s.stopCh: case <-timer.C: - s.warnSyncWriteBacklog(len(req.metrics)) + s.warnSyncWriteBacklog(len(req.metrics) + len(req.availability) + len(req.availabilityDeletes)) } } @@ -1050,6 +1056,14 @@ func (s *Store) processWriteRequests(requests []writeRequest) { if len(combined) > 0 { s.writeBatch(combined) } + for _, req := range requests { + if len(req.availability) > 0 { + s.writeAvailabilityBatch(req.availability) + } + if len(req.availabilityDeletes) > 0 { + s.deleteAvailabilityTargets(req.availabilityDeletes) + } + } for _, done := range doneChans { close(done) } @@ -1171,7 +1185,7 @@ func coalesceMetricBatch(metrics []bufferedMetric) []bufferedMetric { // can commit pending metrics in a single SQLite transaction while preserving // Flush completion barriers. func (s *Store) coalesceQueuedRequests(initial writeRequest) []writeRequest { - if len(initial.metrics) == 0 && initial.done == nil { + if len(initial.metrics) == 0 && len(initial.availability) == 0 && len(initial.availabilityDeletes) == 0 && initial.done == nil { return nil } @@ -1182,7 +1196,7 @@ func (s *Store) coalesceQueuedRequests(initial writeRequest) []writeRequest { if !ok { return combined } - if len(next.metrics) == 0 && next.done == nil { + if len(next.metrics) == 0 && len(next.availability) == 0 && len(next.availabilityDeletes) == 0 && next.done == nil { continue } combined = append(combined, next) @@ -1843,6 +1857,7 @@ func (s *Store) maintenanceWorker() { case <-retentionTicker.C: s.runRetention() + s.runAvailabilityRetention() } } } diff --git a/pkg/metrics/store_query_plan_test.go b/pkg/metrics/store_query_plan_test.go index b25991ae2..1b78a8bc2 100644 --- a/pkg/metrics/store_query_plan_test.go +++ b/pkg/metrics/store_query_plan_test.go @@ -252,6 +252,62 @@ func TestQueryPlansUseIndexes(t *testing.T) { } } +func TestAvailabilityHistoryQueryPlansUseIndexes(t *testing.T) { + db := newPlanTestDB(t) + + tests := []struct { + name string + query string + args []any + wantIndex string + }{ + { + name: "fleet rollup batch", + query: `SELECT target_id, bucket_start, reachable_millis, unreachable_millis, + indeterminate_millis, latency_count, latency_sum, + COALESCE(latency_min, 0), COALESCE(latency_max, 0) + FROM availability_history_buckets + WHERE target_id IN (?, ?, ?) AND tier = ? AND bucket_start >= ? AND bucket_start < ? + ORDER BY target_id, bucket_start`, + args: []any{"target-1", "target-2", "target-3", "minute", int64(0), int64(9999999999)}, + wantIndex: "sqlite_autoindex_availability_history_buckets_1", + }, + { + name: "latest live tail batch", + query: `SELECT target_id, timeline_at_ns, valid_until_ns + FROM availability_observations + WHERE target_id IN (?, ?, ?) + ORDER BY target_id, timeline_at_ns DESC, id DESC`, + args: []any{"target-1", "target-2", "target-3"}, + wantIndex: "idx_availability_observations_target_timeline", + }, + { + name: "revision boundary batch", + query: `SELECT target_id, revision, started_at_ns + FROM availability_revision_boundaries + WHERE target_id IN (?, ?, ?) AND started_at_ns >= ? AND started_at_ns < ? + ORDER BY target_id, started_at_ns`, + args: []any{"target-1", "target-2", "target-3", int64(0), int64(9999999999)}, + wantIndex: "idx_availability_boundaries_target_time", + }, + { + name: "raw retention", + query: `DELETE FROM availability_observations WHERE timeline_at_ns < ?`, + args: []any{int64(9999999999)}, + wantIndex: "idx_availability_observations_retention", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plan := explainQueryPlan(t, db, tt.query, tt.args) + if !strings.Contains(plan, "SEARCH") || !strings.Contains(plan, tt.wantIndex) { + t.Fatalf("expected indexed availability-history search using %s\nPlan:\n%s", tt.wantIndex, plan) + } + }) + } +} + // lineRefersToMetrics returns true if a plan line references the "metrics" // table (not metrics_meta). SQLite may emit "SEARCH metrics USING ...", // "SEARCH TABLE metrics ...", "SCAN metrics ...", or "SCAN TABLE metrics ...". @@ -383,6 +439,52 @@ func newPlanTestDB(t *testing.T) *sql.DB { key TEXT PRIMARY KEY, value TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS availability_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + observation_id TEXT NOT NULL UNIQUE, + target_id TEXT NOT NULL, + config_revision INTEGER NOT NULL, + outcome TEXT NOT NULL, + observed_at_ns INTEGER NOT NULL, + timeline_at_ns INTEGER NOT NULL, + ingested_at_ns INTEGER NOT NULL, + valid_until_ns INTEGER NOT NULL, + execution_source TEXT NOT NULL, + latency_millis INTEGER + ); + CREATE INDEX IF NOT EXISTS idx_availability_observations_target_timeline + ON availability_observations(target_id, timeline_at_ns, id); + CREATE INDEX IF NOT EXISTS idx_availability_observations_retention + ON availability_observations(timeline_at_ns); + + CREATE TABLE IF NOT EXISTS availability_history_buckets ( + target_id TEXT NOT NULL, + tier TEXT NOT NULL, + bucket_start INTEGER NOT NULL, + reachable_millis INTEGER NOT NULL DEFAULT 0, + unreachable_millis INTEGER NOT NULL DEFAULT 0, + indeterminate_millis INTEGER NOT NULL DEFAULT 0, + reachable_count INTEGER NOT NULL DEFAULT 0, + unreachable_count INTEGER NOT NULL DEFAULT 0, + indeterminate_count INTEGER NOT NULL DEFAULT 0, + latency_count INTEGER NOT NULL DEFAULT 0, + latency_sum INTEGER NOT NULL DEFAULT 0, + latency_min INTEGER, + latency_max INTEGER, + PRIMARY KEY(target_id, tier, bucket_start) + ); + CREATE INDEX IF NOT EXISTS idx_availability_buckets_retention + ON availability_history_buckets(tier, bucket_start); + + CREATE TABLE IF NOT EXISTS availability_revision_boundaries ( + target_id TEXT NOT NULL, + revision INTEGER NOT NULL, + started_at_ns INTEGER NOT NULL, + PRIMARY KEY(target_id, revision) + ); + CREATE INDEX IF NOT EXISTS idx_availability_boundaries_target_time + ON availability_revision_boundaries(target_id, started_at_ns); ` if _, err := db.Exec(schema); err != nil { t.Fatalf("create schema: %v", err) @@ -398,6 +500,28 @@ func newPlanTestDB(t *testing.T) *sql.DB { if err != nil { t.Fatalf("seed data: %v", err) } + + targetID := fmt.Sprintf("target-%d", i%10) + _, err = db.Exec(`INSERT INTO availability_observations + (observation_id, target_id, config_revision, outcome, observed_at_ns, timeline_at_ns, + ingested_at_ns, valid_until_ns, execution_source, latency_millis) + VALUES (?, ?, 1, 'reachable', ?, ?, ?, ?, 'local', 12)`, + fmt.Sprintf("observation-%d", i), targetID, int64(1000000+i*5), int64(1000000+i*5), + int64(1000000+i*5), int64(1000060+i*5)) + if err != nil { + t.Fatalf("seed availability observations: %v", err) + } + _, err = db.Exec(`INSERT OR IGNORE INTO availability_history_buckets + (target_id, tier, bucket_start, reachable_millis, latency_count, latency_sum, latency_min, latency_max) + VALUES (?, 'minute', ?, 60000, 1, 12, 12, 12)`, targetID, int64(1000000+i*60)) + if err != nil { + t.Fatalf("seed availability buckets: %v", err) + } + _, err = db.Exec(`INSERT OR IGNORE INTO availability_revision_boundaries + (target_id, revision, started_at_ns) VALUES (?, 1, ?)`, targetID, int64(1000000+i*5)) + if err != nil { + t.Fatalf("seed availability boundaries: %v", err) + } } // Run ANALYZE so the planner has real statistics.