Add availability history and fleet view

This commit is contained in:
Pulse Test
2026-08-30 15:35:10 +01:00
parent a7f1a5e25f
commit 712f1a3bfe
39 changed files with 2843 additions and 242 deletions
+23 -1
View File
@@ -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.
@@ -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
+1 -1
View File
@@ -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": [
@@ -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
@@ -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
@@ -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
@@ -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,
@@ -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
@@ -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
@@ -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
+26 -20
View File
@@ -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"
]
}
@@ -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);
});
});
@@ -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<AvailabilityHistoryResponse> {
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<AvailabilityHistoryResponse>(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),
};
}
}
@@ -32,6 +32,7 @@ export interface AvailabilityProbeStatus {
export interface AvailabilityTarget {
id: string;
configRevision?: number;
name: string;
targetKind?: AvailabilityTargetKind;
address: string;
@@ -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={
<div>
<div class="mb-1.5 text-[10px] font-semibold uppercase tracking-wide text-muted">
Availability presentation
</div>
<FilterSegmentedControl
aria-label="Availability presentation"
value={props.view ?? 'table'}
onChange={(value) => props.onViewChange?.(value as AvailabilityChecksView)}
options={[
{ value: 'table', label: 'Table' },
{ value: 'fleet', label: 'Fleet' },
]}
/>
</div>
}
/>
<Show
@@ -133,190 +200,205 @@ export const AvailabilityChecksTable: Component<{
/>
}
>
<PlatformTableShell
title="Availability checks"
actions={
<div class="flex flex-wrap items-center justify-end gap-2">
<A href={buildAvailabilityTargetAddPath('service')} class={settingsLinkClass}>
<PlusIcon class="h-3.5 w-3.5" />
Add service/device check
</A>
<A href={buildAvailabilitySettingsPath()} class={settingsLinkClass}>
<SettingsIcon class="h-3.5 w-3.5" />
Manage
</A>
</div>
<Show
when={(props.view ?? 'table') === 'table'}
fallback={
<AvailabilityFleetView
resources={orderedChecks()}
historyByTarget={historyByTarget()}
historyLoading={history.loading}
historyError={history()?.error}
probeAgentOptions={props.probeAgentOptions}
onRetryHistory={() => void historyActions.refetch()}
/>
}
tableClass="min-w-full table-fixed text-xs md:min-w-[900px]"
header={
<>
<TableHead
class={`${getPlatformTableHeadClassForKind('name')} platform-table-mobile-w-30 md:w-[20%]`}
>
Check
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('text')} platform-table-mobile-w-15 md:w-[12%]`}
>
Method
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('text')} platform-table-mobile-w-25 md:w-[22%]`}
>
Target
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} platform-table-mobile-w-15 md:w-[12%]`}
>
Result
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} platform-table-mobile-w-15 md:w-[10%]`}
>
<PlatformResponsiveTableLabel compact="Seen" full="Checked" />
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[10%]`}
>
Last healthy
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[8%]`}
>
Failures
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[8%]`}
>
Interval
</TableHead>
</>
}
body={
<PlatformWindowedRows items={orderedChecks} estimatedRowHeight={32}>
{(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);
>
<PlatformTableShell
title="Availability checks"
actions={
<div class="flex flex-wrap items-center justify-end gap-2">
<A href={buildAvailabilityTargetAddPath('service')} class={settingsLinkClass}>
<PlusIcon class="h-3.5 w-3.5" />
Add service/device check
</A>
<A href={buildAvailabilitySettingsPath()} class={settingsLinkClass}>
<SettingsIcon class="h-3.5 w-3.5" />
Manage
</A>
</div>
}
tableClass="min-w-full table-fixed text-xs md:min-w-[900px]"
header={
<>
<TableHead
class={`${getPlatformTableHeadClassForKind('name')} platform-table-mobile-w-30 md:w-[20%]`}
>
Check
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('text')} platform-table-mobile-w-15 md:w-[12%]`}
>
Method
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('text')} platform-table-mobile-w-25 md:w-[22%]`}
>
Target
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} platform-table-mobile-w-15 md:w-[12%]`}
>
Result
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} platform-table-mobile-w-15 md:w-[10%]`}
>
<PlatformResponsiveTableLabel compact="Seen" full="Checked" />
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[10%]`}
>
Last healthy
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[8%]`}
>
Failures
</TableHead>
<TableHead
class={`${getPlatformTableHeadClassForKind('numeric-value')} hidden lg:table-cell lg:w-[8%]`}
>
Interval
</TableHead>
</>
}
body={
<PlatformWindowedRows items={orderedChecks} estimatedRowHeight={32}>
{(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 (
<>
<TableRow
data-availability-check-row={check.id}
class={`${getPlatformResourceDetailRowClass(isExpanded())} text-[11px] sm:text-xs`}
aria-controls={isExpanded() ? detailRowId() : undefined}
aria-expanded={isExpanded() ? 'true' : 'false'}
onClick={() => drawer.toggle(check)}
onKeyDown={drawer.handleActivationKey(check)}
tabIndex={0}
>
<TableCell class={getPlatformTableCellClassForKind('name')}>
<div class="flex min-w-0 items-center gap-2">
<PlatformResourceDetailToggleButton
expanded={isExpanded()}
resourceLabel={check.name}
controlsId={detailRowId()}
onToggle={() => drawer.toggle(check)}
/>
<StatusDot
size="sm"
variant={indicator().variant}
title={indicator().label}
ariaHidden
/>
<span
class="truncate font-semibold text-base-content"
title={check.name}
>
{check.name}
</span>
</div>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('text')} text-base-content`}
return (
<>
<TableRow
data-availability-check-row={check.id}
class={`${getPlatformResourceDetailRowClass(isExpanded())} text-[11px] sm:text-xs`}
aria-controls={isExpanded() ? detailRowId() : undefined}
aria-expanded={isExpanded() ? 'true' : 'false'}
onClick={() => drawer.toggle(check)}
onKeyDown={drawer.handleActivationKey(check)}
tabIndex={0}
>
{method()}
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('text')} text-base-content`}
>
<span class="block truncate" title={target()}>
{target()}
</span>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} text-base-content`}
>
<span class={probe()?.toneClassName ?? ''} title={probe()?.detailLabel}>
{result()}
</span>
<Show when={probeSource()}>
{(sourceLabel) => (
<MetadataBadge
tone="muted"
size="xs"
appearance="outline"
class="mt-0.5 flex"
data-availability-probe-source={availability()?.probeAgentId}
<TableCell class={getPlatformTableCellClassForKind('name')}>
<div class="flex min-w-0 items-center gap-2">
<PlatformResourceDetailToggleButton
expanded={isExpanded()}
resourceLabel={check.name}
controlsId={detailRowId()}
onToggle={() => drawer.toggle(check)}
/>
<StatusDot
size="sm"
variant={indicator().variant}
title={indicator().label}
ariaHidden
/>
<span
class="truncate font-semibold text-base-content"
title={check.name}
>
{sourceLabel()}
</MetadataBadge>
)}
</Show>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} text-base-content`}
>
<PlatformTableRelativeTimeValue
value={availability()?.lastChecked}
emptyText="Not checked"
/>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
<PlatformTableRelativeTimeValue
value={availability()?.lastSuccess}
emptyText="Never"
/>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
{formatFailures(availability())}
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
<PlatformTableDurationValue
seconds={availability()?.pollIntervalSeconds}
/>
</TableCell>
</TableRow>
<PlatformResourceDetailTableRow
resource={check}
open={isExpanded()}
detailRowId={detailRowId()}
colSpan={8}
resolveResourceLabel={resolveResourceLabel}
onClose={() => drawer.close(check)}
/>
</>
);
}}
</PlatformWindowedRows>
}
/>
{check.name}
</span>
</div>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('text')} text-base-content`}
>
{method()}
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('text')} text-base-content`}
>
<span class="block truncate" title={target()}>
{target()}
</span>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} text-base-content`}
>
<span class={probe()?.toneClassName ?? ''} title={probe()?.detailLabel}>
{result()}
</span>
<Show when={probeSource()}>
{(sourceLabel) => (
<MetadataBadge
tone="muted"
size="xs"
appearance="outline"
class="mt-0.5 flex"
data-availability-probe-source={availability()?.probeAgentId}
>
{sourceLabel()}
</MetadataBadge>
)}
</Show>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} text-base-content`}
>
<PlatformTableRelativeTimeValue
value={availability()?.lastChecked}
emptyText="Not checked"
/>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
<PlatformTableRelativeTimeValue
value={availability()?.lastSuccess}
emptyText="Never"
/>
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
{formatFailures(availability())}
</TableCell>
<TableCell
class={`${getPlatformTableCellClassForKind('numeric-value')} hidden text-base-content lg:table-cell`}
>
<PlatformTableDurationValue
seconds={availability()?.pollIntervalSeconds}
/>
</TableCell>
</TableRow>
<PlatformResourceDetailTableRow
resource={check}
open={isExpanded()}
detailRowId={detailRowId()}
colSpan={8}
resolveResourceLabel={resolveResourceLabel}
onClose={() => drawer.close(check)}
/>
</>
);
}}
</PlatformWindowedRows>
}
/>
</Show>
</Show>
</div>
</Show>
@@ -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<AvailabilityFleetHistoryState, number>();
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 (
<div
class="flex h-3 w-full overflow-hidden rounded-sm bg-slate-200 dark:bg-slate-700"
role="img"
aria-label={`24-hour state history: ${description() || 'no observations'}`}
data-testid="availability-state-strip"
>
<For each={states()}>
{(state) => (
<span
class={`min-w-px flex-1 ${statePresentation[state].className}`}
title={statePresentation[state].label}
/>
)}
</For>
</div>
);
};
const AvailabilityLatencyLine: Component<{ buckets: readonly AvailabilityHistoryBucket[] }> = (
props,
) => {
const paths = createMemo(() => latencyPaths(props.buckets));
return (
<Show
when={paths().length > 0}
fallback={
<div class="flex h-7 items-center text-[10px] text-muted">No reachable latency</div>
}
>
<svg
viewBox="0 0 100 26"
preserveAspectRatio="none"
class="h-7 w-full overflow-visible"
role="img"
aria-label={paths()[0]?.label ?? 'Reachable latency'}
data-testid="availability-latency-line"
>
<For each={paths()}>
{(path) => (
<polyline
points={path.path}
fill="none"
stroke="currentColor"
stroke-width="1.75"
vector-effect="non-scaling-stroke"
class="text-sky-500 dark:text-sky-300"
/>
)}
</For>
</svg>
</Show>
);
};
export const AvailabilityFleetView: Component<{
resources: readonly Resource[];
historyByTarget: ReadonlyMap<string, AvailabilityHistoryTarget>;
historyLoading: boolean;
historyError?: string;
probeAgentOptions?: readonly ProbeAgentOption[];
onRetryHistory?: () => void;
}> = (props) => {
const [selectedResource, setSelectedResource] = createSignal<Resource>();
const resolveResourceLabel = (resourceId: string): string | undefined =>
props.resources.find((resource) => resource.id === resourceId)?.name;
return (
<section aria-label="Availability fleet" class="space-y-3">
<div class="flex flex-wrap items-center justify-between gap-2 rounded-md border border-border bg-surface-alt/40 px-3 py-2 text-xs text-muted">
<div class="flex flex-wrap items-center gap-x-3 gap-y-1" aria-label="State history legend">
<For each={Object.values(statePresentation)}>
{(presentation) => (
<span class="inline-flex items-center gap-1.5">
<span
class={`h-2.5 w-2.5 rounded-sm ${presentation.className}`}
aria-hidden="true"
/>
{presentation.label}
</span>
)}
</For>
<span class="text-[10px]">24-hour evidence · reachable latency only</span>
</div>
<Show when={props.historyError}>
<div class="flex items-center gap-2 text-rose-700 dark:text-rose-300">
<span>History unavailable. Current health is unchanged.</span>
<Show when={props.onRetryHistory}>
<Button type="button" size="xs" variant="secondary" onClick={props.onRetryHistory}>
<RefreshCwIcon class="h-3 w-3" />
Retry
</Button>
</Show>
</div>
</Show>
</div>
<div
class="grid grid-cols-1 gap-2 sm:grid-cols-2 xl:grid-cols-3 2xl:grid-cols-4"
data-testid="availability-fleet-grid"
>
<For each={props.resources}>
{(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 (
<button
type="button"
class="group min-w-0 rounded-md border border-border bg-surface p-3 text-left shadow-sm transition hover:border-blue-400/60 hover:bg-surface-hover focus-visible:outline focus-visible:outline-2 focus-visible:outline-offset-2 focus-visible:outline-blue-500"
aria-label={`Open details for ${resource.name}`}
data-availability-fleet-tile={resource.id}
onClick={() => setSelectedResource(resource)}
>
<div class="flex items-start justify-between gap-3">
<div class="min-w-0">
<div class="flex min-w-0 items-center gap-2">
<StatusDot
size="sm"
variant={indicator().variant}
title={indicator().label}
ariaHidden
/>
<span
class="truncate text-sm font-semibold text-base-content"
title={resource.name}
>
{resource.name}
</span>
</div>
<div class="mt-1 truncate text-[11px] text-muted" title={endpoint()}>
{endpoint()}
</div>
</div>
<span class={probe()?.toneClassName ?? 'text-muted'}>
{probe()?.resultLabel ?? indicator().label}
</span>
</div>
<div class="mt-2 flex flex-wrap items-center gap-1.5 text-[10px] text-muted">
<MetadataBadge tone="muted" size="xs" appearance="outline">
{probe()?.methodLabel ?? availability()?.protocol ?? 'Probe'}
</MetadataBadge>
<MetadataBadge tone="muted" size="xs" appearance="outline">
{source()}
</MetadataBadge>
<span>Checked {checked()}</span>
<Show when={latency()}>{(value) => <span>{value()}</span>}</Show>
</div>
<div class="mt-3 space-y-1.5">
<Show
when={!props.historyLoading}
fallback={
<div
class="h-3 animate-pulse rounded-sm bg-surface-hover"
aria-label="Loading history"
/>
}
>
<AvailabilityStateStrip buckets={buckets()} />
</Show>
<AvailabilityLatencyLine buckets={buckets()} />
<div class="flex items-center justify-between gap-2 text-[10px] text-muted">
<span>{availabilityText(history())}</span>
<Show when={history()?.revisionBoundaries?.length}>
<span title="The check configuration changed during this window">
{history()?.revisionBoundaries?.length} revision
{history()?.revisionBoundaries?.length === 1 ? '' : 's'}
</span>
</Show>
</div>
</div>
</button>
);
}}
</For>
</div>
<Show when={selectedResource()}>
{(resource) => (
<ResourceDetailDrawer
resource={resource()}
resolveResourceLabel={resolveResourceLabel}
onClose={() => setSelectedResource(undefined)}
/>
)}
</Show>
</section>
);
};
export default AvailabilityFleetView;
@@ -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<AvailabilityChecksView>(() =>
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() {
<AvailabilityChecksTable
resources={model().availabilityChecks}
probeAgentOptions={buildProbeAgentOptions(model().machines)}
view={availabilityView()}
onViewChange={setAvailabilityView}
externalSearch={machineSearchFilter}
onExternalSearchChange={setMachineSearchFilter}
externalStatus={machineStatusFilter}
onExternalStatusChange={setMachineStatusFilter}
onResetFilters={resetMachineFilters}
emptyIcon={availabilityIcon()}
emptyTitle="No availability checks"
emptyDescription="Add ping, TCP, MQTT, ESPHome, or HTTP checks for devices and services that cannot run Pulse Agent."
@@ -0,0 +1,147 @@
import { cleanup, fireEvent, render, screen } from '@solidjs/testing-library';
import { afterEach, describe, expect, it, vi } from 'vitest';
import type { AvailabilityHistoryTarget } from '@/api/availabilityHistory';
import type { Resource } from '@/types/resource';
import { AvailabilityFleetView } from '../AvailabilityFleetView';
vi.mock('@/components/Infrastructure/ResourceDetailDrawer', () => ({
ResourceDetailDrawer: (props: { resource: Resource; onClose?: () => void }) => (
<div data-testid="resource-detail-drawer" data-resource-id={props.resource.id}>
<button type="button" onClick={props.onClose}>
Close
</button>
</div>
),
}));
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(() => (
<AvailabilityFleetView
resources={resources}
historyByTarget={historyByTarget}
historyLoading={false}
/>
));
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(() => (
<AvailabilityFleetView
resources={[resource(1)]}
historyByTarget={new Map()}
historyLoading={false}
historyError="request failed"
/>
));
expect(
screen.getByText('History unavailable. Current health is unchanged.'),
).toBeInTheDocument();
expect(screen.getByText('History unavailable')).toBeInTheDocument();
expect(screen.getByText('Reachable')).toBeInTheDocument();
});
});
@@ -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');
@@ -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';
+12 -1
View File
@@ -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})
}
@@ -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()
@@ -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
}
@@ -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())
}
}
+1
View File
@@ -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",
+1
View File
@@ -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)))
+1
View File
@@ -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",
+23
View File
@@ -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
}
+34
View File
@@ -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 ",
+8 -5
View File
@@ -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())
+53 -1
View File
@@ -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,
+41 -13
View File
@@ -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,
@@ -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))
+8 -6
View File
@@ -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
+4 -2
View File
@@ -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) {
+843
View File
@@ -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
}
+212
View File
@@ -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, &notNull, &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)
}
}
+23 -8
View File
@@ -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()
}
}
}
+124
View File
@@ -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.