mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-08-29 03:36:55 +00:00
fix(notifications): replace polling with Docker event stream for container lifecycle detection (#588)
* fix(notifications): replace polling with Docker event stream for container lifecycle detection Replaces the 30-second MonitorService crash-detection poll with a causal, per-node Docker events stream. Eliminates false crash alerts on intentional stops (docker stop, compose down, stack restart/update), detects OOM kills as a distinct alert category, and surfaces real crashes in real time. A new DockerEventManager spawns one DockerEventService per local node. Each service consumes the filtered container event stream, classifies die events against recent kill/oom state, and reconciles container state via snapshot diffing on connect and reconnect. Rate limiting, exponential backoff with jitter, and parse-error tolerance keep the stream resilient under load and during daemon interruptions. MonitorService retains host limits, janitor, version check, and stack metric alerts; crash and healthcheck detection move out entirely. * fix(tests): silence require-imports lint in hoisted mock factory
This commit is contained in:
Anso
@@ -0,0 +1,73 @@
|
||||
/**
|
||||
* ContainerLifecycleClassifier
|
||||
*
|
||||
* Pure classification helpers for Docker container lifecycle events. No I/O,
|
||||
* no side effects, no singletons. Consumed by DockerEventService.
|
||||
*
|
||||
* The classifier answers a single question: given a `die` event and the
|
||||
* container's recent lifecycle state, is this exit intentional, a clean exit,
|
||||
* an OOM kill, or a crash worth alerting on?
|
||||
*/
|
||||
|
||||
export type Classification = 'intentional' | 'clean' | 'crash' | 'oom';
|
||||
|
||||
/** Window (ms) after a `kill` event within which a subsequent `die` is considered intentional. */
|
||||
export const INTENTIONAL_KILL_WINDOW_MS = 60_000;
|
||||
|
||||
export interface ContainerLifecycleState {
|
||||
/** Timestamp (ms) of the most recent `kill` event for this container, if any. */
|
||||
lastKillAt?: number;
|
||||
/** True when an `oom` event has been observed and the matching `die` has not yet arrived. */
|
||||
oomPending?: boolean;
|
||||
}
|
||||
|
||||
export interface DieEventInput {
|
||||
/** Time the die event occurred (ms). Typically Date.now() when the event was parsed. */
|
||||
at: number;
|
||||
/** Exit code reported by Docker. May be undefined for malformed events (treated as non-zero). */
|
||||
exitCode: number | undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a die event against the container's current lifecycle state.
|
||||
*
|
||||
* Priority order:
|
||||
* 1. OOM pending → 'oom' (OOM kills are meaningful even if exitCode looks clean)
|
||||
* 2. Recent kill within window → 'intentional'
|
||||
* 3. Exit code 0 → 'clean'
|
||||
* 4. Anything else → 'crash'
|
||||
*/
|
||||
export function classifyDie(
|
||||
input: DieEventInput,
|
||||
state: ContainerLifecycleState,
|
||||
): Classification {
|
||||
if (state.oomPending) return 'oom';
|
||||
|
||||
if (typeof state.lastKillAt === 'number') {
|
||||
// Use absolute age so out-of-order deliveries (kill arrives slightly
|
||||
// after die) still classify as intentional. DockerEventService's 500ms
|
||||
// die grace window makes this realistically bounded.
|
||||
const age = Math.abs(input.at - state.lastKillAt);
|
||||
if (age <= INTENTIONAL_KILL_WINDOW_MS) {
|
||||
return 'intentional';
|
||||
}
|
||||
}
|
||||
|
||||
if (input.exitCode === 0) return 'clean';
|
||||
return 'crash';
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a gap exit discovered during reconciliation (no die event observed
|
||||
* because the stream was disconnected). Uses the container inspect result
|
||||
* rather than event state.
|
||||
*/
|
||||
export function classifyGapExit(inspect: {
|
||||
State?: { OOMKilled?: boolean; ExitCode?: number };
|
||||
}): Classification {
|
||||
const oom = inspect.State?.OOMKilled === true;
|
||||
if (oom) return 'oom';
|
||||
const code = inspect.State?.ExitCode;
|
||||
if (code === 0) return 'clean';
|
||||
return 'crash';
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
import { DatabaseService, Node } from './DatabaseService';
|
||||
import { NodeRegistry } from './NodeRegistry';
|
||||
import { DockerEventService } from './DockerEventService';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
|
||||
/**
|
||||
* DockerEventManager
|
||||
*
|
||||
* Singleton coordinator that owns one DockerEventService per local node.
|
||||
* Spawns services on boot for every existing local node, and reacts to
|
||||
* NodeRegistry 'node-added' / 'node-removed' / 'node-updated' events to
|
||||
* keep the service map in sync with the database.
|
||||
*
|
||||
* Remote nodes self-monitor on their own Sencho instance; this manager does
|
||||
* not subscribe to remote Docker daemons.
|
||||
*/
|
||||
export class DockerEventManager {
|
||||
private static instance: DockerEventManager;
|
||||
private services: Map<number, DockerEventService> = new Map();
|
||||
private started = false;
|
||||
|
||||
private readonly onNodeAdded = (id: number) => { void this.handleNodeAdded(id); };
|
||||
private readonly onNodeRemoved = (id: number) => { this.handleNodeRemoved(id); };
|
||||
private readonly onNodeUpdated = (id: number) => { void this.handleNodeUpdated(id); };
|
||||
|
||||
private constructor() { /* private: use getInstance */ }
|
||||
|
||||
public static getInstance(): DockerEventManager {
|
||||
if (!DockerEventManager.instance) {
|
||||
DockerEventManager.instance = new DockerEventManager();
|
||||
}
|
||||
return DockerEventManager.instance;
|
||||
}
|
||||
|
||||
/** Boot: spawn a DockerEventService for every existing local node. */
|
||||
public async start(): Promise<void> {
|
||||
if (this.started) return;
|
||||
this.started = true;
|
||||
|
||||
const registry = NodeRegistry.getInstance();
|
||||
registry.on('node-added', this.onNodeAdded);
|
||||
registry.on('node-removed', this.onNodeRemoved);
|
||||
registry.on('node-updated', this.onNodeUpdated);
|
||||
|
||||
// Spawn in parallel so one slow node can't block boot for the others.
|
||||
const nodes = DatabaseService.getInstance().getNodes()
|
||||
.filter(n => n.type === 'local' && typeof n.id === 'number');
|
||||
await Promise.all(nodes.map(n => this.spawn(n)));
|
||||
}
|
||||
|
||||
/** Shutdown: stop every service and unsubscribe from registry events. */
|
||||
public stop(): void {
|
||||
if (!this.started) return;
|
||||
this.started = false;
|
||||
|
||||
const registry = NodeRegistry.getInstance();
|
||||
registry.off('node-added', this.onNodeAdded);
|
||||
registry.off('node-removed', this.onNodeRemoved);
|
||||
registry.off('node-updated', this.onNodeUpdated);
|
||||
|
||||
for (const service of this.services.values()) service.shutdown();
|
||||
this.services.clear();
|
||||
}
|
||||
|
||||
/** Aggregated status for diagnostics (e.g. /api/health). */
|
||||
public getStatus(): Array<ReturnType<DockerEventService['getStatus']>> {
|
||||
return Array.from(this.services.values()).map(s => s.getStatus());
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Node lifecycle handlers
|
||||
// ========================================================================
|
||||
|
||||
private async handleNodeAdded(nodeId: number): Promise<void> {
|
||||
if (this.services.has(nodeId)) return;
|
||||
const node = DatabaseService.getInstance().getNode(nodeId);
|
||||
if (!node || node.type !== 'local') return;
|
||||
await this.spawn(node);
|
||||
}
|
||||
|
||||
private handleNodeRemoved(nodeId: number): void {
|
||||
const service = this.services.get(nodeId);
|
||||
if (!service) return;
|
||||
service.shutdown();
|
||||
this.services.delete(nodeId);
|
||||
}
|
||||
|
||||
private async handleNodeUpdated(nodeId: number): Promise<void> {
|
||||
const node = DatabaseService.getInstance().getNode(nodeId);
|
||||
const existing = this.services.get(nodeId);
|
||||
|
||||
// Node became remote (or was deleted): tear down.
|
||||
if (!node || node.type !== 'local') {
|
||||
if (existing) {
|
||||
existing.shutdown();
|
||||
this.services.delete(nodeId);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
// Node is local: ensure a service exists (respawn if missing).
|
||||
if (!existing) {
|
||||
await this.spawn(node);
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Service spawning
|
||||
// ========================================================================
|
||||
|
||||
private async spawn(node: Node): Promise<void> {
|
||||
if (typeof node.id !== 'number') return;
|
||||
if (this.services.has(node.id)) return;
|
||||
|
||||
const service = new DockerEventService(node.id, node.name);
|
||||
this.services.set(node.id, service);
|
||||
|
||||
try {
|
||||
await service.start();
|
||||
} catch (err) {
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventManager] failed to start service for node ${node.name}:`,
|
||||
err instanceof Error ? err.message : err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,621 @@
|
||||
import Docker from 'dockerode';
|
||||
import { NodeRegistry } from './NodeRegistry';
|
||||
import { NotificationService } from './NotificationService';
|
||||
import { DatabaseService } from './DatabaseService';
|
||||
import {
|
||||
classifyDie,
|
||||
classifyGapExit,
|
||||
Classification,
|
||||
ContainerLifecycleState,
|
||||
} from './ContainerLifecycleClassifier';
|
||||
import { isDebugEnabled } from '../utils/debug';
|
||||
|
||||
/**
|
||||
* DockerEventService
|
||||
*
|
||||
* Subscribes to Docker's container event stream for a single local node and
|
||||
* translates causal events (kill / die / oom / health_status) into alerts.
|
||||
*
|
||||
* One instance is spawned per local node by DockerEventManager. Each instance
|
||||
* owns a dedicated Docker client, stream, reconnect timer, and state map - no
|
||||
* shared mutable state between per-node services.
|
||||
*
|
||||
* See docs/features/alerts-notifications.mdx for user-facing behaviour.
|
||||
*/
|
||||
|
||||
/** Grace window after a `die` before classifying, to absorb out-of-order kill events. */
|
||||
const DIE_GRACE_WINDOW_MS = 500;
|
||||
|
||||
/** Max crash alerts emitted per node within RATE_WINDOW_MS. Overflow is batched. */
|
||||
const RATE_LIMIT_MAX = 20;
|
||||
const RATE_WINDOW_MS = 60_000;
|
||||
|
||||
/** Dedup window for repeat crash alerts of the same container. */
|
||||
const CRASH_DEDUP_MS = 60 * 60_000;
|
||||
|
||||
/** Interval for pruning stale container state from memory. */
|
||||
const PRUNE_INTERVAL_MS = 60_000;
|
||||
const STATE_STALE_AFTER_MS = 10 * 60_000;
|
||||
|
||||
/** Parse-error threshold: >N errors per window triggers a single warning alert. */
|
||||
const PARSE_ERROR_THRESHOLD = 10;
|
||||
const PARSE_ERROR_WINDOW_MS = 60_000;
|
||||
|
||||
/** Fraction of exited containers on reconnect that triggers mass-event handling. */
|
||||
const MASS_EVENT_THRESHOLD = 0.2;
|
||||
|
||||
/** Reconnect backoff bounds. */
|
||||
const RECONNECT_BASE_MS = 1_000;
|
||||
const RECONNECT_MAX_MS = 60_000;
|
||||
const RECONNECT_JITTER_MS = 500;
|
||||
|
||||
/** Compose project label key used by docker compose on every container it creates. */
|
||||
const COMPOSE_PROJECT_LABEL = 'com.docker.compose.project';
|
||||
|
||||
/** TTL for the cached global_crash settings flag (sub-second so toggle takes effect quickly). */
|
||||
const SETTINGS_CACHE_MS = 500;
|
||||
|
||||
interface InternalContainerState extends ContainerLifecycleState {
|
||||
name?: string;
|
||||
stackName?: string;
|
||||
lastCrashAlertAt?: number;
|
||||
lastActivityAt: number;
|
||||
}
|
||||
|
||||
interface DockerEventPayload {
|
||||
Type?: string;
|
||||
Action?: string;
|
||||
Actor?: {
|
||||
ID?: string;
|
||||
Attributes?: Record<string, string>;
|
||||
};
|
||||
time?: number;
|
||||
timeNano?: number;
|
||||
}
|
||||
|
||||
type LifecycleStatus = 'disconnected' | 'connecting' | 'connected' | 'stopped';
|
||||
|
||||
export class DockerEventService {
|
||||
private readonly nodeId: number;
|
||||
private readonly nodeName: string;
|
||||
private readonly docker: Docker;
|
||||
private readonly notifier: NotificationService;
|
||||
|
||||
private status: LifecycleStatus = 'disconnected';
|
||||
private stream: NodeJS.ReadableStream | null = null;
|
||||
private reconnectAttempts = 0;
|
||||
private reconnectTimer: NodeJS.Timeout | null = null;
|
||||
private pruneTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
/** Per-container lifecycle state, keyed by Docker container ID. */
|
||||
private containerState: Map<string, InternalContainerState> = new Map();
|
||||
|
||||
/** Pending die timers keyed by container ID (for the 500ms grace window). */
|
||||
private pendingDieTimers: Map<string, NodeJS.Timeout> = new Map();
|
||||
|
||||
/** Rate-limiter bookkeeping. */
|
||||
private rateWindowStart = 0;
|
||||
private rateCount = 0;
|
||||
private suppressedCount = 0;
|
||||
private summaryTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
/** Parse-error tracking for flooded bad payloads. */
|
||||
private parseErrorWindowStart = 0;
|
||||
private parseErrorCount = 0;
|
||||
private parseWarningEmitted = false;
|
||||
|
||||
/** True once we've completed the initial boot reconciliation. */
|
||||
private bootReconciled = false;
|
||||
|
||||
/** IDs of containers that were exited at the last known-good moment. */
|
||||
private exitedBaseline: Set<string> = new Set();
|
||||
|
||||
/** Whether we've already emitted the one-time "lost connection" warning. */
|
||||
private disconnectedNoticeEmitted = false;
|
||||
|
||||
/** Cache for the global_crash toggle to avoid a DB read per event. */
|
||||
private crashAlertsCache: { value: boolean; at: number } | null = null;
|
||||
|
||||
constructor(nodeId: number, nodeName: string) {
|
||||
this.nodeId = nodeId;
|
||||
this.nodeName = nodeName;
|
||||
this.docker = NodeRegistry.getInstance().getDocker(nodeId);
|
||||
this.notifier = NotificationService.getInstance();
|
||||
}
|
||||
|
||||
/** Open the event stream and begin consuming events. Safe to call once. */
|
||||
public async start(): Promise<void> {
|
||||
if (this.status !== 'disconnected') return;
|
||||
this.pruneTimer = setInterval(() => this.pruneStaleState(), PRUNE_INTERVAL_MS);
|
||||
await this.connect();
|
||||
}
|
||||
|
||||
/** Close the stream, cancel timers, and clear state. */
|
||||
public shutdown(): void {
|
||||
this.status = 'stopped';
|
||||
this.clearReconnectTimer();
|
||||
if (this.pruneTimer) {
|
||||
clearInterval(this.pruneTimer);
|
||||
this.pruneTimer = null;
|
||||
}
|
||||
if (this.summaryTimer) {
|
||||
clearTimeout(this.summaryTimer);
|
||||
this.summaryTimer = null;
|
||||
}
|
||||
for (const timer of this.pendingDieTimers.values()) clearTimeout(timer);
|
||||
this.pendingDieTimers.clear();
|
||||
this.detachStream();
|
||||
this.containerState.clear();
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Connection lifecycle
|
||||
// ========================================================================
|
||||
|
||||
private async connect(): Promise<void> {
|
||||
if (this.status === 'stopped') return;
|
||||
this.status = 'connecting';
|
||||
try {
|
||||
const stream = await this.docker.getEvents({
|
||||
filters: { type: ['container'] },
|
||||
}) as unknown as NodeJS.ReadableStream;
|
||||
|
||||
this.stream = stream;
|
||||
this.status = 'connected';
|
||||
|
||||
if (this.disconnectedNoticeEmitted) {
|
||||
await this.emitInfo(`Reconnected to Docker daemon.`);
|
||||
this.disconnectedNoticeEmitted = false;
|
||||
}
|
||||
|
||||
this.reconnectAttempts = 0;
|
||||
this.attachStreamHandlers(stream);
|
||||
|
||||
await this.reconcile();
|
||||
} catch (error) {
|
||||
this.handleDisconnect(error);
|
||||
}
|
||||
}
|
||||
|
||||
private attachStreamHandlers(stream: NodeJS.ReadableStream): void {
|
||||
let buffer = '';
|
||||
stream.on('data', (chunk: Buffer) => {
|
||||
buffer += chunk.toString('utf8');
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() ?? '';
|
||||
for (const line of lines) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) continue;
|
||||
this.handleRawEvent(trimmed);
|
||||
}
|
||||
});
|
||||
stream.on('error', (err) => this.handleDisconnect(err));
|
||||
stream.on('end', () => this.handleDisconnect(new Error('Event stream ended')));
|
||||
stream.on('close', () => this.handleDisconnect(new Error('Event stream closed')));
|
||||
}
|
||||
|
||||
private detachStream(): void {
|
||||
const s = this.stream;
|
||||
this.stream = null;
|
||||
if (!s) return;
|
||||
try {
|
||||
s.removeAllListeners();
|
||||
const destroyable = s as unknown as { destroy?: () => void };
|
||||
destroyable.destroy?.();
|
||||
} catch {
|
||||
/* noop */
|
||||
}
|
||||
}
|
||||
|
||||
private handleDisconnect(error: unknown): void {
|
||||
if (this.status === 'stopped') return;
|
||||
this.detachStream();
|
||||
this.status = 'disconnected';
|
||||
|
||||
if (!this.disconnectedNoticeEmitted) {
|
||||
this.disconnectedNoticeEmitted = true;
|
||||
void this.emitWarning(`Lost connection to Docker daemon; monitoring paused.`);
|
||||
}
|
||||
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventService:${this.nodeName}] disconnected:`,
|
||||
error instanceof Error ? error.message : error);
|
||||
}
|
||||
this.scheduleReconnect();
|
||||
}
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
if (this.status === 'stopped') return;
|
||||
this.clearReconnectTimer();
|
||||
const attempt = this.reconnectAttempts;
|
||||
const base = Math.min(RECONNECT_MAX_MS, RECONNECT_BASE_MS * Math.pow(2, attempt));
|
||||
const jitter = Math.floor(Math.random() * RECONNECT_JITTER_MS);
|
||||
const delay = base + jitter;
|
||||
this.reconnectAttempts += 1;
|
||||
this.reconnectTimer = setTimeout(() => {
|
||||
this.reconnectTimer = null;
|
||||
void this.connect();
|
||||
}, delay);
|
||||
}
|
||||
|
||||
private clearReconnectTimer(): void {
|
||||
if (this.reconnectTimer) {
|
||||
clearTimeout(this.reconnectTimer);
|
||||
this.reconnectTimer = null;
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Reconciliation
|
||||
// ========================================================================
|
||||
|
||||
/**
|
||||
* Snapshot all containers on connect. On first boot, record the baseline
|
||||
* silently. On subsequent reconnects, treat newly-exited containers as
|
||||
* gap exits and classify them (or batch as a mass event).
|
||||
*/
|
||||
private async reconcile(): Promise<void> {
|
||||
let containers;
|
||||
try {
|
||||
containers = await this.docker.listContainers({ all: true });
|
||||
} catch (err) {
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventService:${this.nodeName}] reconcile list failed:`,
|
||||
err instanceof Error ? err.message : err);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const exitedNow = new Set(
|
||||
containers.filter(c => c.State === 'exited').map(c => c.Id)
|
||||
);
|
||||
|
||||
if (!this.bootReconciled) {
|
||||
this.exitedBaseline = exitedNow;
|
||||
this.bootReconciled = true;
|
||||
return;
|
||||
}
|
||||
|
||||
const newlyExited = [...exitedNow].filter(id => !this.exitedBaseline.has(id));
|
||||
const totalKnown = Math.max(1, containers.length);
|
||||
const exitRatio = newlyExited.length / totalKnown;
|
||||
|
||||
if (newlyExited.length === 0) {
|
||||
this.exitedBaseline = exitedNow;
|
||||
return;
|
||||
}
|
||||
|
||||
if (exitRatio >= MASS_EVENT_THRESHOLD) {
|
||||
await this.emitInfo(
|
||||
`Docker daemon interruption detected: ${newlyExited.length} containers exited during connection gap.`
|
||||
);
|
||||
} else {
|
||||
// Inspect + classify in parallel. Below the mass-event threshold
|
||||
// newlyExited is small by definition, so unbounded concurrency is fine.
|
||||
await Promise.all(newlyExited.map(id => this.classifyGap(id)));
|
||||
}
|
||||
|
||||
this.exitedBaseline = exitedNow;
|
||||
}
|
||||
|
||||
private async classifyGap(containerId: string): Promise<void> {
|
||||
try {
|
||||
const inspect = await this.docker.getContainer(containerId).inspect();
|
||||
const classification = classifyGapExit({ State: inspect.State });
|
||||
if (classification === 'clean' || classification === 'intentional') return;
|
||||
|
||||
const name = inspect.Name?.replace(/^\//, '') ?? containerId.slice(0, 12);
|
||||
const stackName = inspect.Config?.Labels?.[COMPOSE_PROJECT_LABEL];
|
||||
const exitCode = inspect.State?.ExitCode ?? 0;
|
||||
|
||||
// Gap exits have no in-memory state, so there's no dedup to bump.
|
||||
await this.emitClassification(classification, null, { name, exitCode, stackName });
|
||||
} catch (err) {
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventService:${this.nodeName}] gap inspect failed:`,
|
||||
err instanceof Error ? err.message : err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Event handling
|
||||
// ========================================================================
|
||||
|
||||
private handleRawEvent(line: string): void {
|
||||
let payload: DockerEventPayload;
|
||||
try {
|
||||
payload = JSON.parse(line);
|
||||
} catch {
|
||||
this.trackParseError();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
this.handleEvent(payload);
|
||||
} catch (err) {
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventService:${this.nodeName}] event handler threw:`,
|
||||
err instanceof Error ? err.message : err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private handleEvent(event: DockerEventPayload): void {
|
||||
if (event.Type !== 'container') return;
|
||||
const action = event.Action ?? '';
|
||||
const id = event.Actor?.ID;
|
||||
if (!id) return;
|
||||
|
||||
// Normalize: `health_status: unhealthy` -> base action
|
||||
const baseAction = action.startsWith('health_status') ? 'health_status' : action;
|
||||
|
||||
switch (baseAction) {
|
||||
case 'kill':
|
||||
return this.onKill(id, event);
|
||||
case 'die':
|
||||
return this.onDie(id, event);
|
||||
case 'oom':
|
||||
return this.onOom(id);
|
||||
case 'health_status':
|
||||
return this.onHealthStatus(id, action, event);
|
||||
case 'start':
|
||||
return this.onStart(id);
|
||||
case 'destroy':
|
||||
return this.onDestroy(id);
|
||||
}
|
||||
}
|
||||
|
||||
private onKill(id: string, event: DockerEventPayload): void {
|
||||
const state = this.getOrCreateState(id, event);
|
||||
state.lastKillAt = this.eventTimeMs(event);
|
||||
state.lastActivityAt = Date.now();
|
||||
}
|
||||
|
||||
private onDie(id: string, event: DockerEventPayload): void {
|
||||
// Defer classification to absorb out-of-order kill events.
|
||||
const existing = this.pendingDieTimers.get(id);
|
||||
if (existing) clearTimeout(existing);
|
||||
const timer = setTimeout(() => {
|
||||
this.pendingDieTimers.delete(id);
|
||||
this.classifyDie(id, event);
|
||||
}, DIE_GRACE_WINDOW_MS);
|
||||
this.pendingDieTimers.set(id, timer);
|
||||
}
|
||||
|
||||
private onOom(id: string): void {
|
||||
const state = this.getOrCreateState(id);
|
||||
state.oomPending = true;
|
||||
state.lastActivityAt = Date.now();
|
||||
}
|
||||
|
||||
private onHealthStatus(id: string, action: string, event: DockerEventPayload): void {
|
||||
if (!action.includes('unhealthy')) return;
|
||||
const state = this.getOrCreateState(id, event);
|
||||
state.lastActivityAt = Date.now();
|
||||
if (!this.isCrashAlertsEnabled()) return;
|
||||
const name = state.name ?? id.slice(0, 12);
|
||||
const stackName = state.stackName;
|
||||
void this.emitError(
|
||||
`Healthcheck failed: ${name} is unhealthy.`,
|
||||
stackName,
|
||||
);
|
||||
}
|
||||
|
||||
private onStart(id: string): void {
|
||||
const state = this.containerState.get(id);
|
||||
if (!state) return;
|
||||
// Container came back: clear transient flags but keep identity.
|
||||
state.lastKillAt = undefined;
|
||||
state.oomPending = undefined;
|
||||
state.lastCrashAlertAt = undefined;
|
||||
state.lastActivityAt = Date.now();
|
||||
}
|
||||
|
||||
private onDestroy(id: string): void {
|
||||
this.containerState.delete(id);
|
||||
const pending = this.pendingDieTimers.get(id);
|
||||
if (pending) {
|
||||
clearTimeout(pending);
|
||||
this.pendingDieTimers.delete(id);
|
||||
}
|
||||
}
|
||||
|
||||
private classifyDie(id: string, event: DockerEventPayload): void {
|
||||
const state = this.getOrCreateState(id, event);
|
||||
const exitCodeStr = event.Actor?.Attributes?.exitCode;
|
||||
const parsedExit = exitCodeStr !== undefined ? parseInt(exitCodeStr, 10) : undefined;
|
||||
const exitCode = Number.isFinite(parsedExit) ? (parsedExit as number) : undefined;
|
||||
const now = Date.now();
|
||||
|
||||
const classification = classifyDie(
|
||||
{ at: this.eventTimeMs(event), exitCode },
|
||||
{ lastKillAt: state.lastKillAt, oomPending: state.oomPending },
|
||||
);
|
||||
|
||||
// Die arrived: clear the oom flag regardless (we've now used it).
|
||||
state.oomPending = undefined;
|
||||
state.lastActivityAt = now;
|
||||
|
||||
if (classification === 'intentional' || classification === 'clean') return;
|
||||
|
||||
// Dedup: skip if we already alerted on this container within the window.
|
||||
if (state.lastCrashAlertAt && now - state.lastCrashAlertAt < CRASH_DEDUP_MS) {
|
||||
return;
|
||||
}
|
||||
|
||||
void this.emitClassification(classification, state, {
|
||||
name: state.name ?? id.slice(0, 12),
|
||||
exitCode: exitCode ?? 0,
|
||||
stackName: state.stackName,
|
||||
});
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Alert emission + rate limiting
|
||||
// ========================================================================
|
||||
|
||||
private async emitClassification(
|
||||
classification: Classification,
|
||||
state: InternalContainerState | null,
|
||||
info: { name: string; exitCode: number; stackName?: string },
|
||||
): Promise<void> {
|
||||
// Respect the existing global crash-alerts toggle so users who have
|
||||
// disabled these notifications in Settings remain opted out.
|
||||
if (!this.isCrashAlertsEnabled()) return;
|
||||
|
||||
const message = classification === 'oom'
|
||||
? `Container OOM Kill: ${info.name} was killed by the OOM killer (out of memory).`
|
||||
: `Container Crash Detected: ${info.name} exited unexpectedly (Code: ${info.exitCode}).`;
|
||||
|
||||
if (!this.consumeRateToken()) {
|
||||
this.suppressedCount += 1;
|
||||
this.scheduleSummary();
|
||||
return;
|
||||
}
|
||||
|
||||
// Stamp the dedup clock only after the alert is actually dispatched, so
|
||||
// rate-suppressed alerts don't silently lock out the next real crash.
|
||||
if (state) state.lastCrashAlertAt = Date.now();
|
||||
|
||||
await this.emitError(message, info.stackName);
|
||||
}
|
||||
|
||||
private isCrashAlertsEnabled(): boolean {
|
||||
const now = Date.now();
|
||||
const cached = this.crashAlertsCache;
|
||||
if (cached && now - cached.at < SETTINGS_CACHE_MS) return cached.value;
|
||||
let value = false;
|
||||
try {
|
||||
const settings = DatabaseService.getInstance().getGlobalSettings();
|
||||
value = settings['global_crash'] === '1';
|
||||
} catch (err) {
|
||||
// Default-deny on settings lookup failure: don't spam users if the
|
||||
// DB is temporarily unavailable.
|
||||
if (isDebugEnabled()) {
|
||||
console.log(`[DockerEventService:${this.nodeName}] settings lookup failed:`,
|
||||
err instanceof Error ? err.message : err);
|
||||
}
|
||||
}
|
||||
this.crashAlertsCache = { value, at: now };
|
||||
return value;
|
||||
}
|
||||
|
||||
/** Return true if an alert can be emitted now. Side effect: increments counters. */
|
||||
private consumeRateToken(): boolean {
|
||||
const now = Date.now();
|
||||
if (now - this.rateWindowStart >= RATE_WINDOW_MS) {
|
||||
this.rateWindowStart = now;
|
||||
this.rateCount = 0;
|
||||
}
|
||||
if (this.rateCount >= RATE_LIMIT_MAX) return false;
|
||||
this.rateCount += 1;
|
||||
return true;
|
||||
}
|
||||
|
||||
private scheduleSummary(): void {
|
||||
if (this.summaryTimer) return;
|
||||
const remaining = RATE_WINDOW_MS - (Date.now() - this.rateWindowStart);
|
||||
const delay = Math.max(1_000, remaining);
|
||||
this.summaryTimer = setTimeout(() => {
|
||||
const count = this.suppressedCount;
|
||||
this.summaryTimer = null;
|
||||
this.suppressedCount = 0;
|
||||
if (count > 0) {
|
||||
void this.emitWarning(
|
||||
`${count} additional containers crashed in the last minute.`,
|
||||
);
|
||||
}
|
||||
}, delay);
|
||||
}
|
||||
|
||||
private trackParseError(): void {
|
||||
const now = Date.now();
|
||||
if (now - this.parseErrorWindowStart >= PARSE_ERROR_WINDOW_MS) {
|
||||
this.parseErrorWindowStart = now;
|
||||
this.parseErrorCount = 0;
|
||||
this.parseWarningEmitted = false;
|
||||
}
|
||||
this.parseErrorCount += 1;
|
||||
if (this.parseErrorCount > PARSE_ERROR_THRESHOLD && !this.parseWarningEmitted) {
|
||||
this.parseWarningEmitted = true;
|
||||
void this.emitWarning(
|
||||
`Received malformed Docker event payloads. Monitoring continues but some events may be skipped.`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// State + helpers
|
||||
// ========================================================================
|
||||
|
||||
private getOrCreateState(id: string, event?: DockerEventPayload): InternalContainerState {
|
||||
let state = this.containerState.get(id);
|
||||
if (!state) {
|
||||
state = { lastActivityAt: Date.now() };
|
||||
this.containerState.set(id, state);
|
||||
}
|
||||
if (event) {
|
||||
const attrs = event.Actor?.Attributes ?? {};
|
||||
if (attrs.name && !state.name) state.name = attrs.name;
|
||||
const project = attrs[COMPOSE_PROJECT_LABEL];
|
||||
if (project && !state.stackName) state.stackName = project;
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
private eventTimeMs(event: DockerEventPayload): number {
|
||||
if (typeof event.timeNano === 'number') return Math.floor(event.timeNano / 1_000_000);
|
||||
if (typeof event.time === 'number') return event.time * 1000;
|
||||
return Date.now();
|
||||
}
|
||||
|
||||
private pruneStaleState(): void {
|
||||
if (this.containerState.size === 0) return;
|
||||
const cutoff = Date.now() - STATE_STALE_AFTER_MS;
|
||||
for (const [id, state] of this.containerState) {
|
||||
if (state.lastActivityAt < cutoff) {
|
||||
this.containerState.delete(id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Notification wrappers (prefix with node name for multi-node clarity)
|
||||
// ========================================================================
|
||||
|
||||
private async emitError(message: string, stackName?: string): Promise<void> {
|
||||
return this.notifier.dispatchAlert('error', this.prefix(message), stackName);
|
||||
}
|
||||
|
||||
private async emitWarning(message: string, stackName?: string): Promise<void> {
|
||||
return this.notifier.dispatchAlert('warning', this.prefix(message), stackName);
|
||||
}
|
||||
|
||||
private async emitInfo(message: string, stackName?: string): Promise<void> {
|
||||
return this.notifier.dispatchAlert('info', this.prefix(message), stackName);
|
||||
}
|
||||
|
||||
private prefix(message: string): string {
|
||||
return `[Node: ${this.nodeName}] ${message}`;
|
||||
}
|
||||
|
||||
// ========================================================================
|
||||
// Diagnostics
|
||||
// ========================================================================
|
||||
|
||||
public getStatus(): {
|
||||
nodeId: number;
|
||||
nodeName: string;
|
||||
status: LifecycleStatus;
|
||||
reconnectAttempts: number;
|
||||
trackedContainers: number;
|
||||
} {
|
||||
return {
|
||||
nodeId: this.nodeId,
|
||||
nodeName: this.nodeName,
|
||||
status: this.status,
|
||||
reconnectAttempts: this.reconnectAttempts,
|
||||
trackedContainers: this.containerState.size,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -69,10 +69,9 @@ export class MonitorService {
|
||||
// key: rule_id, value: AlertState
|
||||
private activeBreaches = new Map<number, AlertState>();
|
||||
|
||||
// Track containers that have already been alerted as crashed to avoid
|
||||
// duplicate alerts. key: containerId, value: timestamp when alerted.
|
||||
private alertedCrashes = new Map<string, number>();
|
||||
private static readonly CRASH_ALERT_TTL_MS = 60 * 60 * 1000; // 1 hour
|
||||
// Crash and healthcheck detection live in DockerEventService (event-driven,
|
||||
// causal classification). MonitorService no longer polls for container
|
||||
// exits; see backend/src/services/DockerEventService.ts.
|
||||
|
||||
// Sencho version check cooldown (6 hours between external API calls)
|
||||
private lastVersionCheckAt = 0;
|
||||
@@ -126,7 +125,6 @@ export class MonitorService {
|
||||
}
|
||||
|
||||
private async evaluateGlobalSettings(settings: Record<string, string>) {
|
||||
const notifier = NotificationService.getInstance();
|
||||
const HOST_ALERT_COOLDOWN_MS = 5 * 60 * 1000; // 5 minutes between repeat alerts
|
||||
|
||||
// 1. Host Limits
|
||||
@@ -160,61 +158,9 @@ export class MonitorService {
|
||||
console.error('Error checking host limits in watchdog', e);
|
||||
}
|
||||
|
||||
// 2. Global Crash Detect
|
||||
if (settings['global_crash'] === '1') {
|
||||
// Prune expired entries from the crash tracker
|
||||
const now = Date.now();
|
||||
for (const [id, ts] of this.alertedCrashes) {
|
||||
if (now - ts > MonitorService.CRASH_ALERT_TTL_MS) this.alertedCrashes.delete(id);
|
||||
}
|
||||
|
||||
try {
|
||||
const nodes = DatabaseService.getInstance().getNodes();
|
||||
const runningIds = new Set<string>();
|
||||
|
||||
for (const node of nodes) {
|
||||
if (!node.id) continue;
|
||||
// Remote nodes run their own MonitorService locally
|
||||
if (node.type === 'remote') continue;
|
||||
try {
|
||||
const docker = DockerController.getInstance(node.id);
|
||||
const containers = await docker.getAllContainers();
|
||||
for (const c of containers) {
|
||||
if (c.State === 'running') {
|
||||
runningIds.add(c.Id);
|
||||
continue;
|
||||
}
|
||||
// Skip containers already alerted
|
||||
if (this.alertedCrashes.has(c.Id)) continue;
|
||||
|
||||
const containerStack = c.Labels?.['com.docker.compose.project'] || undefined;
|
||||
|
||||
if (c.State === 'exited') {
|
||||
const match = c.Status.match(/Exited \((\d+)\)/i);
|
||||
const exitCode = match ? parseInt(match[1], 10) : null;
|
||||
const intentionalExitCodes = [0, 137, 143, 255];
|
||||
if (exitCode !== null && !intentionalExitCodes.includes(exitCode)) {
|
||||
await notifier.dispatchAlert('error', `[Node: ${node.name}] Container Crash Detected: ${c.Names[0]} exited unexpectedly (Code: ${exitCode}).`, containerStack);
|
||||
this.alertedCrashes.set(c.Id, now);
|
||||
}
|
||||
} else if (String(c.Status).includes('unhealthy')) {
|
||||
await notifier.dispatchAlert('error', `[Node: ${node.name}] Healthcheck Failed: Container ${c.Names[0]} is unhealthy.`, containerStack);
|
||||
this.alertedCrashes.set(c.Id, now);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
console.error(`Error checking crashes on node ${node.name}`, err);
|
||||
}
|
||||
}
|
||||
|
||||
// Clear crash tracking for containers that are running again
|
||||
for (const id of this.alertedCrashes.keys()) {
|
||||
if (runningIds.has(id)) this.alertedCrashes.delete(id);
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('Error checking global crashes', e);
|
||||
}
|
||||
}
|
||||
// 2. (Removed) Container crash + healthcheck detection moved to
|
||||
// DockerEventService: event-driven, causal, distinguishes
|
||||
// intentional stops from real crashes, detects OOM kills.
|
||||
|
||||
// 3. Docker Janitor Check
|
||||
try {
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import Docker from 'dockerode';
|
||||
import axios from 'axios';
|
||||
import { EventEmitter } from 'events';
|
||||
import { DatabaseService, Node } from './DatabaseService';
|
||||
import { fetchRemoteMeta } from './CapabilityRegistry';
|
||||
|
||||
@@ -10,12 +11,23 @@ import { fetchRemoteMeta } from './CapabilityRegistry';
|
||||
* - Local nodes: direct Docker socket connection via Dockerode (unchanged)
|
||||
* - Remote nodes: HTTP/WS proxy to a remote Sencho instance (api_url + api_token)
|
||||
* No direct Docker TCP connections are made for remote nodes.
|
||||
*
|
||||
* Extends EventEmitter so subscribers (e.g. DockerEventManager) can react to
|
||||
* node lifecycle changes. Emits:
|
||||
* - 'node-added' (nodeId: number) after a node is created
|
||||
* - 'node-removed' (nodeId: number) after a node is deleted
|
||||
* - 'node-updated' (nodeId: number) after a node is updated (type may change)
|
||||
* Route handlers in index.ts are responsible for calling the notify* helpers.
|
||||
*/
|
||||
export class NodeRegistry {
|
||||
export class NodeRegistry extends EventEmitter {
|
||||
private static instance: NodeRegistry;
|
||||
private connections: Map<number, Docker> = new Map();
|
||||
|
||||
private constructor() { }
|
||||
private constructor() {
|
||||
super();
|
||||
// Raise the default listener cap (10) so future subscribers do not trip a warning.
|
||||
this.setMaxListeners(50);
|
||||
}
|
||||
|
||||
public static getInstance(): NodeRegistry {
|
||||
if (!NodeRegistry.instance) {
|
||||
@@ -210,6 +222,31 @@ export class NodeRegistry {
|
||||
this.connections.delete(nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit 'node-added' for subscribers (e.g. DockerEventManager).
|
||||
* Call this from the POST /api/nodes route after the DB insert succeeds.
|
||||
*/
|
||||
public notifyNodeAdded(nodeId: number): void {
|
||||
this.emit('node-added', nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit 'node-removed' for subscribers.
|
||||
* Call this from the DELETE /api/nodes/:id route after the DB delete succeeds.
|
||||
*/
|
||||
public notifyNodeRemoved(nodeId: number): void {
|
||||
this.emit('node-removed', nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit 'node-updated' for subscribers. Type changes (local<->remote) are
|
||||
* handled downstream by tearing down and respawning the subscription.
|
||||
* Call this from the PUT /api/nodes/:id route after the DB update succeeds.
|
||||
*/
|
||||
public notifyNodeUpdated(nodeId: number): void {
|
||||
this.emit('node-updated', nodeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all cached connections.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user