diff --git a/scripts/fabric-health-check.sh b/scripts/fabric-health-check.sh new file mode 100644 index 0000000..bc65291 --- /dev/null +++ b/scripts/fabric-health-check.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# FABRIC systemd health check script +# Called by systemd's WatchdogSec mechanism. +# Exits non-zero if the health endpoint reports unhealthy. + +PORT="${FABRIC_PORT:-3000}" +MAX_RETRIES=3 +RETRY_INTERVAL=1 + +for i in $(seq 1 $MAX_RETRIES); do + STATUS=$(curl -sf -o /dev/null -w '%{http_code}' "http://localhost:${PORT}/api/health" 2>/dev/null) + if [ "$STATUS" = "200" ]; then + exit 0 + fi + if [ "$i" -lt "$MAX_RETRIES" ]; then + sleep $RETRY_INTERVAL + fi +done + +echo "FABRIC health check failed: HTTP $STATUS after $MAX_RETRIES attempts" >&2 +exit 1 diff --git a/src/serverMetrics.ts b/src/serverMetrics.ts new file mode 100644 index 0000000..8a09980 --- /dev/null +++ b/src/serverMetrics.ts @@ -0,0 +1,113 @@ +/** + * FABRIC Server Metrics + * + * Collects and exposes internal metrics for /api/health and /api/metrics endpoints. + */ + +import { VERSION } from './index.js'; + +export interface ServerMetricsSnapshot { + status: string; + uptime_sec: number; + version: string; + event_count: number; + ingest_rate_per_sec: number; + ws_clients: number; + tailer_files_watched: number; + dedup_dropped: number; + process_resident_memory_bytes: number; +} + +export class ServerMetrics { + private startTime = Date.now(); + private eventTimestamps: number[] = []; + private _wsClients = 0; + private _tailerFilesWatched = 0; + private _eventCount = 0; + private _dedupDropped = 0; + + recordEvent(): void { + this._eventCount++; + this.eventTimestamps.push(Date.now()); + } + + set wsClients(count: number) { + this._wsClients = count; + } + + set tailerFilesWatched(count: number) { + this._tailerFilesWatched = count; + } + + set dedupDropped(count: number) { + this._dedupDropped = count; + } + + set eventCount(count: number) { + this._eventCount = count; + } + + reset(): void { + this.startTime = Date.now(); + this.eventTimestamps = []; + this._wsClients = 0; + this._tailerFilesWatched = 0; + this._eventCount = 0; + this._dedupDropped = 0; + } + + private ingestRate(): number { + const now = Date.now(); + // Keep only last 60s of timestamps + const cutoff = now - 60_000; + this.eventTimestamps = this.eventTimestamps.filter(t => t >= cutoff); + + if (this.eventTimestamps.length < 2) return 0; + + const spanSec = (now - this.eventTimestamps[0]) / 1000; + if (spanSec < 0.001) return 0; + + return this.eventTimestamps.length / spanSec; + } + + snapshot(): ServerMetricsSnapshot { + const rss = process.memoryUsage().rss; + return { + status: 'ok', + uptime_sec: Math.round((Date.now() - this.startTime) / 1000), + version: VERSION, + event_count: this._eventCount, + ingest_rate_per_sec: Math.round(this.ingestRate() * 100) / 100, + ws_clients: this._wsClients, + tailer_files_watched: this._tailerFilesWatched, + dedup_dropped: this._dedupDropped, + process_resident_memory_bytes: rss, + }; + } + + /** Format snapshot as Prometheus text exposition format. */ + toPrometheus(snap: ServerMetricsSnapshot): string { + const lines: string[] = []; + const metric = (name: string, type: string, help: string, value: number | string, labels?: string) => { + lines.push(`# HELP fabric_${name} ${help}`); + lines.push(`# TYPE fabric_${name} ${type}`); + if (labels) { + lines.push(`fabric_${name}{${labels}} ${value}`); + } else { + lines.push(`fabric_${name} ${value}`); + } + }; + + metric('status', 'gauge', 'Server status (1=ok)', snap.status === 'ok' ? 1 : 0); + metric('uptime_seconds', 'gauge', 'Server uptime in seconds', snap.uptime_sec); + metric('info', 'gauge', 'Build info', 1, `version="${snap.version}"`); + metric('event_count', 'gauge', 'Total events in store', snap.event_count); + metric('ingest_rate_per_second', 'gauge', 'Events ingested per second (60s window)', snap.ingest_rate_per_sec); + metric('websocket_clients', 'gauge', 'Connected WebSocket clients', snap.ws_clients); + metric('tailer_files_watched', 'gauge', 'Log files being watched', snap.tailer_files_watched); + metric('dedup_dropped_total', 'counter', 'Total duplicate events dropped', snap.dedup_dropped); + metric('process_resident_memory_bytes', 'gauge', 'Process RSS in bytes', snap.process_resident_memory_bytes); + + return lines.join('\n') + '\n'; + } +} diff --git a/src/web/server.ts b/src/web/server.ts index 0b325bb..4a2c768 100644 --- a/src/web/server.ts +++ b/src/web/server.ts @@ -9,6 +9,7 @@ import { createServer, Server as HttpServer } from 'http'; import { EventEmitter } from 'events'; import { join, dirname } from 'path'; import { fileURLToPath } from 'url'; +import { createSocket } from 'dgram'; import { WebSocketServer, WebSocket } from 'ws'; import { LogEvent, EventFilter, CrossReferenceEntityType, CrossReferenceRelationship, DagOptions, BeadStatus } from '../types.js'; import { InMemoryEventStore } from '../store.js'; @@ -26,6 +27,21 @@ const MAX_BATCH_SIZE = 100; const __dirname = dirname(fileURLToPath(import.meta.url)); +/** Send a systemd sd_notify message via the NOTIFY_SOCKET Unix datagram socket. */ +function sdNotify(state: string): void { + const socketPath = process.env.NOTIFY_SOCKET; + if (!socketPath) return; + try { + const client = createSocket('unix_dgram'); + const msg = Buffer.from(state); + // Abstract sockets start with '@' in systemd notation; replace with '\0' + const addr = socketPath.startsWith('@') ? '\0' + socketPath.slice(1) : socketPath; + client.send(msg, 0, msg.length, addr, () => client.close()); + } catch { + // Never crash the server due to a notify failure + } +} + export interface WebServerOptions { port: number; logPath: string; @@ -679,6 +695,17 @@ export function createWebServer(options: WebServerOptions): WebServer { ); } console.log('Press Ctrl+C to stop'); + + // Notify systemd that the service is ready (Type=notify) + sdNotify('READY=1\nSTATUS=FABRIC running\n'); + + // Watchdog keepalives: ping at half the configured interval + const watchdogUsec = parseInt(process.env.WATCHDOG_USEC ?? '0', 10); + if (watchdogUsec > 0) { + const intervalMs = Math.floor(watchdogUsec / 2 / 1000); + setInterval(() => sdNotify('WATCHDOG=1'), intervalMs); + } + emitter.emit('start'); });