diff --git a/lib/collectors/scheduler.ts b/lib/collectors/scheduler.ts index cafe4e0..11ad5db 100644 --- a/lib/collectors/scheduler.ts +++ b/lib/collectors/scheduler.ts @@ -7,6 +7,9 @@ import { collectSystem } from "./system"; import { collectHttp } from "./http"; import { collectService } from "./service"; import type { WidgetResult } from "@/lib/types/widget-result"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("scheduler"); const httpLimit = pLimit(8); @@ -122,6 +125,12 @@ class CollectorScheduler { if (existing) clearInterval(existing.timer); + log.debug(existing ? "job replaced" : "job added", { + widgetId: instance.id, + type: instance.widget.type, + intervalMs, + }); + const run = () => this.runJob(instance.id); const timer = setInterval(run, intervalMs); this.jobs.set(instance.id, { widget: instance.widget, intervalMs, timer, consecutiveFailures: 0 }); @@ -130,16 +139,20 @@ class CollectorScheduler { for (const [id, job] of this.jobs) { if (!seen.has(id)) { + log.debug("job removed", { widgetId: id, type: job.widget.type }); clearInterval(job.timer); this.jobs.delete(id); this.lastResults.delete(id); } } + + log.info("reconciled", { jobs: this.jobs.size }); } stop(): void { for (const job of this.jobs.values()) clearInterval(job.timer); this.jobs.clear(); + log.info("scheduler stopped"); } private async runJob(id: string): Promise { @@ -150,10 +163,37 @@ class CollectorScheduler { try { result = await collect(job.widget); if (result.type === "http") { + const wasFailing = job.consecutiveFailures > 0; job.consecutiveFailures = result.data.up ? 0 : job.consecutiveFailures + 1; result = { type: "http", data: { ...result.data, consecutiveFailures: job.consecutiveFailures } }; + if (!result.data.up) { + log.warn("http check failed", { + widgetId: id, + url: (job.widget as { url?: string }).url, + statusCode: result.data.statusCode, + attempt: job.consecutiveFailures, + }); + } else if (wasFailing) { + log.info("http check recovered", { widgetId: id, url: (job.widget as { url?: string }).url }); + } + } else { + if (job.consecutiveFailures > 0) { + log.info("collector recovered", { + widgetId: id, + type: job.widget.type, + afterFailures: job.consecutiveFailures, + }); + } + job.consecutiveFailures = 0; } } catch (err) { + job.consecutiveFailures += 1; + log.error("collector failed", { + widgetId: id, + type: job.widget.type, + attempt: job.consecutiveFailures, + error: (err as Error).message, + }); result = { type: "error", message: (err as Error).message }; } diff --git a/lib/collectors/service.ts b/lib/collectors/service.ts index bfa058a..aa24efc 100644 --- a/lib/collectors/service.ts +++ b/lib/collectors/service.ts @@ -8,6 +8,9 @@ import { collectN8n } from "./services/n8n"; import { collectUmami } from "./services/umami"; import { collectHeadscale } from "./services/headscale"; import { collectTraefikStat } from "./services/traefik"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("service"); function collectServiceStat(widget: ServiceWidget): Promise<{ stats: ServiceStat[] }> { switch (widget.service) { @@ -34,10 +37,11 @@ export async function collectService(widget: ServiceWidget): Promise ({ - stats: [] as ServiceStat[], - statError: (err as Error).message, - })); + const statPromise = collectServiceStat(widget).catch((err) => { + const message = (err as Error).message; + log.warn("service API call failed", { service: widget.service, containerName: widget.containerName, error: message }); + return { stats: [] as ServiceStat[], statError: message }; + }); const [docker, statResult] = await Promise.all([dockerPromise, statPromise]); return { docker, ...statResult }; diff --git a/lib/config/effective.ts b/lib/config/effective.ts index cf5fb0b..700b1e9 100644 --- a/lib/config/effective.ts +++ b/lib/config/effective.ts @@ -1,6 +1,9 @@ import { configStore } from "./loader"; import { discoverDockerWidgets } from "@/lib/discovery/traefik-labels"; import type { Config, Widget } from "./schema"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("discovery"); const DISCOVERY_INTERVAL_MS = 60_000; @@ -53,11 +56,15 @@ class EffectiveConfigStore { try { const next = await discoverDockerWidgets(configStore.get()); if (!sameDiscoveredSet(this.discovered, next)) { + log.info("discovered widget set changed", { + previous: this.discovered.length, + current: next.length, + }); this.discovered = next; this.emit(); } } catch (err) { - console.error(`[discovery] ${(err as Error).message}`); + log.error("discovery scan failed", { error: (err as Error).message }); } } diff --git a/lib/config/loader.ts b/lib/config/loader.ts index b995521..02fb7de 100644 --- a/lib/config/loader.ts +++ b/lib/config/loader.ts @@ -5,11 +5,18 @@ import chokidar, { type FSWatcher } from "chokidar"; import { parse as parseYaml } from "yaml"; import { configSchema, type Config } from "./schema"; import { interpolateEnv } from "./interpolate"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("config"); const CONFIG_DIR = path.join(process.cwd(), "config"); const CONFIG_PATH = path.join(CONFIG_DIR, "config.yml"); const ENV_PATH = path.join(CONFIG_DIR, ".env"); +function widgetCount(config: Config): number { + return config.groups.reduce((sum, group) => sum + group.widgets.length, 0); +} + export class ConfigError extends Error {} function loadEnv(): Record { @@ -62,6 +69,7 @@ class ConfigStore { load(): Config { this.current = readConfig(); + log.info("config loaded", { groups: this.current.groups.length, widgets: widgetCount(this.current) }); return this.current; } @@ -85,7 +93,11 @@ class ConfigStore { awaitWriteFinish: { stabilityThreshold: 200, pollInterval: 50 }, ignoreInitial: true, }); - this.watcher.on("all", () => this.reload()); + this.watcher.on("all", (event) => { + log.debug("watched file changed", { event }); + this.reload(); + }); + log.info("watching config files", { paths: [CONFIG_PATH, ENV_PATH] }); } stop(): void { @@ -97,10 +109,11 @@ class ConfigStore { try { const next = readConfig(); this.current = next; + log.info("config reloaded", { groups: next.groups.length, widgets: widgetCount(next) }); for (const listener of this.listeners) listener(next); } catch (err) { const configError = err instanceof ConfigError ? err : new ConfigError((err as Error).message); - console.error(`[config] ${configError.message}`); + log.error("config reload failed", { error: configError.message }); for (const listener of this.errorListeners) listener(configError); } } diff --git a/lib/logger.ts b/lib/logger.ts new file mode 100644 index 0000000..927df22 --- /dev/null +++ b/lib/logger.ts @@ -0,0 +1,47 @@ +type LogLevel = "debug" | "info" | "warn" | "error"; +type LogFields = Record; + +const LEVEL_ORDER: Record = { debug: 0, info: 1, warn: 2, error: 3 }; + +function resolveMinLevel(): LogLevel { + const configured = process.env.LOG_LEVEL?.toLowerCase(); + if (configured === "debug" || configured === "info" || configured === "warn" || configured === "error") { + return configured; + } + return process.env.NODE_ENV === "production" ? "info" : "debug"; +} + +const MIN_LEVEL = resolveMinLevel(); + +function formatFields(fields?: LogFields): string { + if (!fields) return ""; + const parts = Object.entries(fields) + .filter(([, value]) => value !== undefined) + .map(([key, value]) => `${key}=${typeof value === "string" ? value : JSON.stringify(value)}`); + return parts.length > 0 ? ` ${parts.join(" ")}` : ""; +} + +function write(level: LogLevel, scope: string, message: string, fields?: LogFields): void { + if (LEVEL_ORDER[level] < LEVEL_ORDER[MIN_LEVEL]) return; + const line = `${new Date().toISOString()} ${level.toUpperCase().padEnd(5)} [${scope}] ${message}${formatFields(fields)}`; + if (level === "error") console.error(line); + else if (level === "warn") console.warn(line); + else console.log(line); +} + +export interface Logger { + debug(message: string, fields?: LogFields): void; + info(message: string, fields?: LogFields): void; + warn(message: string, fields?: LogFields): void; + error(message: string, fields?: LogFields): void; +} + +/** Scoped logger. `scope` is the bracketed tag, e.g. createLogger("scheduler") -> "[scheduler]". */ +export function createLogger(scope: string): Logger { + return { + debug: (message, fields) => write("debug", scope, message, fields), + info: (message, fields) => write("info", scope, message, fields), + warn: (message, fields) => write("warn", scope, message, fields), + error: (message, fields) => write("error", scope, message, fields), + }; +} diff --git a/lib/ws/server.ts b/lib/ws/server.ts index e9098b6..ef7a543 100644 --- a/lib/ws/server.ts +++ b/lib/ws/server.ts @@ -6,6 +6,9 @@ import { effectiveConfigStore } from "@/lib/config/effective"; import { toPublicConfig } from "@/lib/config/public"; import type { Config } from "@/lib/config/schema"; import type { WidgetResult } from "@/lib/types/widget-result"; +import { createLogger } from "@/lib/logger"; + +const log = createLogger("ws"); type Envelope = | { topic: string; type: "result"; ts: number; data: WidgetResult } @@ -22,6 +25,8 @@ export function attachWebSocketServer(httpServer: HttpServer): WebSocketServer { const wss = new WebSocketServer({ noServer: true }); const subscriptions = new Map>(); const allSockets = new Set(); + const connectionIds = new WeakMap(); + let nextConnectionId = 1; function subscribe(socket: WebSocket, widgetId: string): void { let set = subscriptions.get(widgetId); @@ -46,12 +51,16 @@ export function attachWebSocketServer(httpServer: HttpServer): WebSocketServer { } wss.on("connection", (socket) => { + const connectionId = nextConnectionId++; + connectionIds.set(socket, connectionId); allSockets.add(socket); + log.info("client connected", { connectionId, clients: allSockets.size }); try { const config = effectiveConfigStore.get(); send(socket, { topic: "config", type: "config:update", ts: Date.now(), data: toPublicConfig(config) }); - } catch { + } catch (err) { + log.debug("no valid config to send on connect", { connectionId, error: (err as Error).message }); // no valid config loaded yet; client keeps its SSR-seeded config until one arrives } @@ -59,18 +68,27 @@ export function attachWebSocketServer(httpServer: HttpServer): WebSocketServer { let message: { action?: string; widgetId?: string }; try { message = JSON.parse(raw.toString()); - } catch { + } catch (err) { + log.warn("received malformed message", { connectionId, error: (err as Error).message }); return; } if (message.action === "subscribe" && typeof message.widgetId === "string") { + log.debug("client subscribed", { connectionId, widgetId: message.widgetId }); subscribe(socket, message.widgetId); } else if (message.action === "unsubscribe" && typeof message.widgetId === "string") { + log.debug("client unsubscribed", { connectionId, widgetId: message.widgetId }); unsubscribe(socket, message.widgetId); } }); - socket.on("close", () => cleanupSocket(socket)); - socket.on("error", () => cleanupSocket(socket)); + socket.on("close", (code) => { + log.info("client disconnected", { connectionId, code, clients: allSockets.size - 1 }); + cleanupSocket(socket); + }); + socket.on("error", (err) => { + log.error("socket error", { connectionId, error: err.message }); + cleanupSocket(socket); + }); }); collectorScheduler.onResult((widgetId, result) => { diff --git a/server.ts b/server.ts index 0e44d0e..5a60cd0 100644 --- a/server.ts +++ b/server.ts @@ -4,6 +4,9 @@ import { attachWebSocketServer } from "./lib/ws/server"; import { configStore } from "./lib/config/loader"; import { effectiveConfigStore } from "./lib/config/effective"; import { collectorScheduler } from "./lib/collectors/scheduler"; +import { createLogger } from "./lib/logger"; + +const log = createLogger("server"); const port = Number(process.env.PORT ?? 3000); const hostname = process.env.HOSTNAME ?? "0.0.0.0"; @@ -14,6 +17,7 @@ const handle = app.getRequestHandler(); async function main(): Promise { await app.prepare(); + log.info("next.js prepared", { dev }); const upgradeHandler = app.getUpgradeHandler(); configStore.load(); @@ -21,6 +25,7 @@ async function main(): Promise { await effectiveConfigStore.start(); collectorScheduler.start(effectiveConfigStore.get()); effectiveConfigStore.onUpdate((config) => collectorScheduler.reconcile(config)); + log.info("collector scheduler started"); const httpServer = createServer((req, res) => { handle(req, res); @@ -39,23 +44,30 @@ async function main(): Promise { void upgradeHandler(req, socket, head); }); - function shutdown(): void { + function shutdown(signal: NodeJS.Signals): void { + log.info("shutting down", { signal }); collectorScheduler.stop(); effectiveConfigStore.stop(); configStore.stop(); - httpServer.close(() => process.exit(0)); - setTimeout(() => process.exit(0), 5_000).unref(); + httpServer.close(() => { + log.info("http server closed"); + process.exit(0); + }); + setTimeout(() => { + log.warn("shutdown timed out, forcing exit"); + process.exit(0); + }, 5_000).unref(); } - process.on("SIGTERM", shutdown); - process.on("SIGINT", shutdown); + process.on("SIGTERM", () => shutdown("SIGTERM")); + process.on("SIGINT", () => shutdown("SIGINT")); httpServer.listen(port, () => { - console.log(`> PulseNode listening on http://${hostname}:${port}`); + log.info("listening", { url: `http://${hostname}:${port}` }); }); } main().catch((err) => { - console.error(err); + log.error("fatal startup error", { error: (err as Error).message, stack: (err as Error).stack }); process.exit(1); });