Repository Public
1/**2 * Cluster telemetry. `ClusterSource` is the adapter seam; `AgentdSource` polls agentd.3 * The last good snapshot is kept with its timestamp: when agentd stops answering, data ages4 * into `stale` / UNKNOWN instead of being shown as current. Without agentd the overview is5 * limited to the host itself, measured by the runtime.6 */7import type { FastifyBaseLogger } from 'fastify';8import type { ClusterNode, ClusterOverview } from '@cr/shared';9import type { ServerConfig } from '../../config.js';10import type { Db } from '../../db.js';11import type { RuntimeClient } from '../../runtime-client.js';12import { STALE_MS, mapAgentd } from './mapping.js';1314export interface ClusterSource {15 /** Raw agentd `/api/cluster` payload. Rejects on any failure. */16 fetch(): Promise<unknown>;17}1819export class AgentdSource implements ClusterSource {20 constructor(21 private readonly url: string,22 private readonly token: string | null,23 private readonly fetchImpl: typeof fetch,24 ) {}25 async fetch(): Promise<unknown> {26 const res = await this.fetchImpl(`${this.url}/api/cluster`, {27 headers: this.token ? { authorization: `Bearer ${this.token}` } : {},28 signal: AbortSignal.timeout(8000),29 });30 if (!res.ok) throw new Error(`agentd responded ${res.status}`);31 return res.json();32 }33}3435export interface HostMetrics {36 collectedAtMs: number;37 hostname: string;38 cpuPct: number | null;39 load: number[];40 memUsedMb: number | null;41 memTotalMb: number | null;42 diskFreeGb: number | null;43 diskTotalGb: number | null;44 uptimeSec: number | null;45}46const shortHost = (h: string): string => (h.split('.')[0] ?? h).slice(0, 40);4748export class ClusterService {49 private snapshot: { payload: unknown; atMs: number } | null = null;50 private lastError: string | null = null;51 private lastPersistMs = 0;52 private hostname: string | null = null;53 private timer: NodeJS.Timeout | null = null;5455 constructor(56 private readonly d: { config: ServerConfig; db: Db; rt: RuntimeClient; log: FastifyBaseLogger; source: ClusterSource | null; now: () => number },57 ) {}5859 get configured(): boolean {60 return this.d.source !== null;61 }62 /** Freshness of the agentd link for /system/status. */63 status(): { configured: boolean; lastGoodAt: Date | null; fresh: boolean; error: string | null } {64 const at = this.snapshot?.atMs ?? null;65 return { configured: this.configured, lastGoodAt: at ? new Date(at) : null, fresh: at !== null && this.d.now() - at <= STALE_MS, error: this.lastError };66 }6768 start(): void {69 if (!this.d.source || this.timer) return;70 void this.poll();71 this.timer = setInterval(() => void this.poll(), this.d.config.clusterPollMs);72 this.timer.unref();73 }74 stop(): void {75 if (this.timer) clearInterval(this.timer);76 this.timer = null;77 }7879 async poll(): Promise<void> {80 if (!this.d.source) return;81 try {82 const payload = await this.d.source.fetch();83 this.snapshot = { payload, atMs: this.d.now() };84 this.lastError = null;85 } catch (err) {86 this.lastError = (err as Error).name === 'TimeoutError' ? 'timeout' : 'unreachable';87 return;88 }89 if (this.d.now() - this.lastPersistMs >= 60_000) {90 this.lastPersistMs = this.d.now();91 await this.persist(await this.overview()).catch((err: unknown) => this.d.log.warn({ err }, 'cluster persist failed'));92 }93 }9495 private async hostAlias(): Promise<string | null> {96 if (this.d.config.nodeAlias) return this.d.config.nodeAlias;97 if (!this.hostname) {98 const m = await this.hostMetrics();99 if (m?.hostname) this.hostname = shortHost(m.hostname);100 }101 return this.hostname;102 }103 hostMetrics(): Promise<HostMetrics | null> {104 return this.d.rt.json<HostMetrics>('GET', '/host/metrics', undefined, { timeoutMs: 5000 }).catch(() => null);105 }106107 async overview(): Promise<ClusterOverview> {108 const now = this.d.now();109 if (this.d.source) {110 if (!this.snapshot) {111 return { source: 'agentd', collectedAt: null, health: 'UNKNOWN', nodes: [], edges: [], totals: { nodes: 0, online: 0, cores: 0, memoryGb: 0 } };112 }113 return mapAgentd(this.snapshot.payload, { nowMs: now, fetchedAtMs: this.snapshot.atMs, execHosts: this.d.config.execHosts, hostAlias: await this.hostAlias() });114 }115 // No agentd: only what the runtime measures on this host. Cluster health stays UNKNOWN.116 const m = await this.hostMetrics();117 const nodes: ClusterNode[] = [];118 if (m && typeof m.hostname === 'string') {119 const stale = now - m.collectedAtMs > STALE_MS;120 nodes.push({121 id: this.d.config.nodeAlias ?? shortHost(m.hostname), model: null, chip: null, kind: 'mac', role: 'host', cpuCores: null, gpuCores: null,122 memoryMb: m.memTotalMb ?? null, os: null, health: stale ? 'UNKNOWN' : 'ONLINE', metricsAt: new Date(m.collectedAtMs).toISOString(), stale,123 cpuPct: m.cpuPct ?? null, memUsedMb: m.memUsedMb ?? null, diskFreeGb: m.diskFreeGb ?? null, diskTotalGb: m.diskTotalGb ?? null,124 load1: m.load?.[0] ?? null, apps: [], execAllowed: false, isHost: true,125 });126 }127 return {128 source: 'none', collectedAt: m ? new Date(m.collectedAtMs).toISOString() : null, health: 'UNKNOWN', nodes, edges: [],129 totals: { nodes: nodes.length, online: nodes.filter((n) => n.health === 'ONLINE').length, cores: 0, memoryGb: Math.round((m?.memTotalMb ?? 0) / 1024) },130 };131 }132133 /** Inventory upsert + one downsampled metrics row per node (called at most once a minute). */134 async persist(o: ClusterOverview): Promise<void> {135 for (const n of o.nodes) {136 await this.d.db.q(137 `INSERT INTO cluster_nodes (id, model, chip, kind, role, cpu_cores, gpu_cores, memory_mb, os, exec_allowed, last_seen_at)138 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, CASE WHEN $11 THEN now() END)139 ON CONFLICT (id) DO UPDATE SET model = excluded.model, chip = excluded.chip, kind = excluded.kind, role = excluded.role,140 cpu_cores = excluded.cpu_cores, gpu_cores = excluded.gpu_cores, memory_mb = excluded.memory_mb, os = excluded.os,141 exec_allowed = excluded.exec_allowed, last_seen_at = coalesce(excluded.last_seen_at, cluster_nodes.last_seen_at)`,142 [n.id, n.model, n.chip, n.kind, n.role, n.cpuCores, n.gpuCores, n.memoryMb === null ? null : Math.round(n.memoryMb), n.os, n.execAllowed, n.health === 'ONLINE'],143 );144 if (n.metricsAt && !n.stale) {145 await this.d.db.q(146 'INSERT INTO node_metrics (node_id, at, online, cpu_pct, mem_used_mb, disk_free_gb, load1) VALUES ($1, $2, $3, $4, $5, $6, $7)',147 [n.id, new Date(n.metricsAt), n.health === 'ONLINE', n.cpuPct, n.memUsedMb, n.diskFreeGb, n.load1],148 );149 }150 }151 }152}153