mirror of
https://github.com/khoaliber/dockhand.git
synced 2026-03-03 21:19:06 +00:00
Initial commit
This commit is contained in:
Jarek Krochmalski
446
lib/server/subprocesses/event-subprocess.ts
Normal file
446
lib/server/subprocesses/event-subprocess.ts
Normal file
@@ -0,0 +1,446 @@
|
||||
/**
|
||||
* Event Collection Subprocess
|
||||
*
|
||||
* Runs as a separate process via Bun.spawn to collect Docker container events
|
||||
* without blocking the main HTTP thread.
|
||||
*
|
||||
* Communication with main process via IPC (process.send).
|
||||
*/
|
||||
|
||||
import { getEnvironments, type ContainerEventAction } from '../db';
|
||||
import { getDockerEvents } from '../docker';
|
||||
import type { MainProcessCommand } from '../subprocess-manager';
|
||||
|
||||
// Reconnection settings
|
||||
const RECONNECT_DELAY = 5000; // 5 seconds
|
||||
const MAX_RECONNECT_DELAY = 60000; // 1 minute max
|
||||
|
||||
// Track environment online status for notifications
|
||||
// Only send notifications on status CHANGES, not on every reconnect attempt
|
||||
const environmentOnlineStatus: Map<number, boolean> = new Map();
|
||||
|
||||
// Active collectors per environment
|
||||
const collectors: Map<number, AbortController> = new Map();
|
||||
|
||||
// Recent event cache for deduplication (key: timeNano-containerId-action)
|
||||
const recentEvents: Map<string, number> = new Map();
|
||||
const DEDUP_WINDOW_MS = 5000; // 5 second window for deduplication
|
||||
const CACHE_CLEANUP_INTERVAL_MS = 30000; // Clean up cache every 30 seconds
|
||||
|
||||
let cacheCleanupInterval: ReturnType<typeof setInterval> | null = null;
|
||||
let isShuttingDown = false;
|
||||
|
||||
// Actions we care about for container activity
|
||||
const CONTAINER_ACTIONS: ContainerEventAction[] = [
|
||||
'create',
|
||||
'start',
|
||||
'stop',
|
||||
'die',
|
||||
'kill',
|
||||
'restart',
|
||||
'pause',
|
||||
'unpause',
|
||||
'destroy',
|
||||
'rename',
|
||||
'update',
|
||||
'oom',
|
||||
'health_status'
|
||||
];
|
||||
|
||||
// Scanner image patterns to exclude from events
|
||||
const SCANNER_IMAGE_PATTERNS = [
|
||||
'anchore/grype',
|
||||
'aquasec/trivy',
|
||||
'ghcr.io/anchore/grype',
|
||||
'ghcr.io/aquasecurity/trivy'
|
||||
];
|
||||
|
||||
// Container name patterns to exclude from events
|
||||
const EXCLUDED_CONTAINER_PREFIXES = ['dockhand-browse-'];
|
||||
|
||||
/**
|
||||
* Send message to main process
|
||||
*/
|
||||
function send(message: any): void {
|
||||
if (process.send) {
|
||||
process.send(message);
|
||||
}
|
||||
}
|
||||
|
||||
function isScannerContainer(image: string | null | undefined): boolean {
|
||||
if (!image) return false;
|
||||
const lowerImage = image.toLowerCase();
|
||||
return SCANNER_IMAGE_PATTERNS.some((pattern) => lowerImage.includes(pattern.toLowerCase()));
|
||||
}
|
||||
|
||||
function isExcludedContainer(containerName: string | null | undefined): boolean {
|
||||
if (!containerName) return false;
|
||||
return EXCLUDED_CONTAINER_PREFIXES.some((prefix) => containerName.startsWith(prefix));
|
||||
}
|
||||
|
||||
/**
|
||||
* Update environment online status and notify main process on change
|
||||
*/
|
||||
function updateEnvironmentStatus(
|
||||
envId: number,
|
||||
envName: string,
|
||||
isOnline: boolean,
|
||||
errorMessage?: string
|
||||
) {
|
||||
const previousStatus = environmentOnlineStatus.get(envId);
|
||||
|
||||
// Only send notification on status CHANGE (not on first connection or repeated failures)
|
||||
if (previousStatus !== undefined && previousStatus !== isOnline) {
|
||||
send({
|
||||
type: 'env_status',
|
||||
envId,
|
||||
envName,
|
||||
online: isOnline,
|
||||
error: errorMessage
|
||||
});
|
||||
}
|
||||
|
||||
environmentOnlineStatus.set(envId, isOnline);
|
||||
}
|
||||
|
||||
interface DockerEvent {
|
||||
Type: string;
|
||||
Action: string;
|
||||
Actor: {
|
||||
ID: string;
|
||||
Attributes: Record<string, string>;
|
||||
};
|
||||
time: number;
|
||||
timeNano: number;
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up old entries from the deduplication cache
|
||||
*/
|
||||
function cleanupRecentEvents() {
|
||||
const now = Date.now();
|
||||
for (const [key, timestamp] of recentEvents.entries()) {
|
||||
if (now - timestamp > DEDUP_WINDOW_MS) {
|
||||
recentEvents.delete(key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Process a Docker event
|
||||
*/
|
||||
function processEvent(event: DockerEvent, envId: number) {
|
||||
// Only process container events
|
||||
if (event.Type !== 'container') return;
|
||||
|
||||
// Map Docker action to our action type
|
||||
const action = event.Action.split(':')[0] as ContainerEventAction;
|
||||
|
||||
// Skip actions we don't care about
|
||||
if (!CONTAINER_ACTIONS.includes(action)) return;
|
||||
|
||||
const containerId = event.Actor?.ID;
|
||||
const containerName = event.Actor?.Attributes?.name;
|
||||
const image = event.Actor?.Attributes?.image;
|
||||
|
||||
if (!containerId) return;
|
||||
|
||||
// Skip scanner containers (Trivy, Grype)
|
||||
if (isScannerContainer(image)) return;
|
||||
|
||||
// Skip internal Dockhand containers (volume browser helpers)
|
||||
if (isExcludedContainer(containerName)) return;
|
||||
|
||||
// Deduplicate events
|
||||
const dedupKey = `${envId}-${event.timeNano}-${containerId}-${action}`;
|
||||
if (recentEvents.has(dedupKey)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Mark as processed
|
||||
recentEvents.set(dedupKey, Date.now());
|
||||
|
||||
// Clean up if cache gets too large
|
||||
if (recentEvents.size > 200) {
|
||||
cleanupRecentEvents();
|
||||
}
|
||||
|
||||
// Convert Unix nanosecond timestamp to ISO string
|
||||
const timestamp = new Date(Math.floor(event.timeNano / 1000000)).toISOString();
|
||||
|
||||
// Prepare notification data
|
||||
const actionLabel = action.charAt(0).toUpperCase() + action.slice(1);
|
||||
const containerLabel = containerName || containerId.substring(0, 12);
|
||||
const notificationType =
|
||||
action === 'die' || action === 'kill' || action === 'oom'
|
||||
? 'error'
|
||||
: action === 'stop'
|
||||
? 'warning'
|
||||
: action === 'start'
|
||||
? 'success'
|
||||
: 'info';
|
||||
|
||||
// Send event to main process for DB save and SSE broadcast
|
||||
send({
|
||||
type: 'container_event',
|
||||
event: {
|
||||
environmentId: envId,
|
||||
containerId: containerId,
|
||||
containerName: containerName || null,
|
||||
image: image || null,
|
||||
action,
|
||||
actorAttributes: event.Actor?.Attributes || null,
|
||||
timestamp
|
||||
},
|
||||
notification: {
|
||||
action,
|
||||
title: `Container ${actionLabel}`,
|
||||
message: `Container "${containerLabel}" ${action}${image ? ` (${image})` : ''}`,
|
||||
notificationType,
|
||||
image
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Start collecting events for a specific environment
|
||||
*/
|
||||
async function startEnvironmentCollector(envId: number, envName: string) {
|
||||
// Stop existing collector if any
|
||||
stopEnvironmentCollector(envId);
|
||||
|
||||
const controller = new AbortController();
|
||||
collectors.set(envId, controller);
|
||||
|
||||
let reconnectDelay = RECONNECT_DELAY;
|
||||
|
||||
const connect = async () => {
|
||||
if (controller.signal.aborted || isShuttingDown) return;
|
||||
|
||||
let reader: ReadableStreamDefaultReader<Uint8Array> | null = null;
|
||||
|
||||
try {
|
||||
console.log(
|
||||
`[EventSubprocess] Connecting to Docker events for ${envName} (env ${envId})...`
|
||||
);
|
||||
|
||||
const eventStream = await getDockerEvents({ type: ['container'] }, envId);
|
||||
|
||||
if (!eventStream) {
|
||||
console.error(`[EventSubprocess] Failed to get event stream for ${envName}`);
|
||||
updateEnvironmentStatus(envId, envName, false, 'Failed to connect to Docker');
|
||||
scheduleReconnect();
|
||||
return;
|
||||
}
|
||||
|
||||
// Reset reconnect delay on successful connection
|
||||
reconnectDelay = RECONNECT_DELAY;
|
||||
console.log(`[EventSubprocess] Connected to Docker events for ${envName}`);
|
||||
|
||||
updateEnvironmentStatus(envId, envName, true);
|
||||
|
||||
reader = eventStream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
|
||||
try {
|
||||
while (!controller.signal.aborted && !isShuttingDown) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() || '';
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.trim()) {
|
||||
try {
|
||||
const event = JSON.parse(line) as DockerEvent;
|
||||
processEvent(event, envId);
|
||||
} catch {
|
||||
// Ignore parse errors for partial chunks
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (error: any) {
|
||||
if (!controller.signal.aborted && !isShuttingDown) {
|
||||
if (error.name !== 'AbortError') {
|
||||
console.error(`[EventSubprocess] Stream error for ${envName}:`, error.message);
|
||||
updateEnvironmentStatus(envId, envName, false, error.message);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
if (reader) {
|
||||
try {
|
||||
reader.releaseLock();
|
||||
} catch {
|
||||
// Reader already released or stream closed - ignore
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Connection closed, reconnect
|
||||
if (!controller.signal.aborted && !isShuttingDown) {
|
||||
scheduleReconnect();
|
||||
}
|
||||
} catch (error: any) {
|
||||
if (reader) {
|
||||
try {
|
||||
reader.releaseLock();
|
||||
} catch {
|
||||
// Reader already released or stream closed - ignore
|
||||
}
|
||||
}
|
||||
|
||||
if (!controller.signal.aborted && !isShuttingDown && error.name !== 'AbortError') {
|
||||
console.error(`[EventSubprocess] Connection error for ${envName}:`, error.message);
|
||||
updateEnvironmentStatus(envId, envName, false, error.message);
|
||||
}
|
||||
|
||||
if (!controller.signal.aborted && !isShuttingDown) {
|
||||
scheduleReconnect();
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
const scheduleReconnect = () => {
|
||||
if (controller.signal.aborted || isShuttingDown) return;
|
||||
|
||||
console.log(`[EventSubprocess] Reconnecting to ${envName} in ${reconnectDelay / 1000}s...`);
|
||||
setTimeout(() => {
|
||||
if (!controller.signal.aborted && !isShuttingDown) {
|
||||
connect();
|
||||
}
|
||||
}, reconnectDelay);
|
||||
|
||||
// Exponential backoff
|
||||
reconnectDelay = Math.min(reconnectDelay * 2, MAX_RECONNECT_DELAY);
|
||||
};
|
||||
|
||||
// Start the connection
|
||||
connect();
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop collecting events for a specific environment
|
||||
*/
|
||||
function stopEnvironmentCollector(envId: number) {
|
||||
const controller = collectors.get(envId);
|
||||
if (controller) {
|
||||
controller.abort();
|
||||
collectors.delete(envId);
|
||||
environmentOnlineStatus.delete(envId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh collectors when environments change
|
||||
*/
|
||||
async function refreshEventCollectors() {
|
||||
if (isShuttingDown) return;
|
||||
|
||||
try {
|
||||
const environments = await getEnvironments();
|
||||
|
||||
// Filter: only collect for environments with activity enabled AND not Hawser Edge
|
||||
const activeEnvIds = new Set(
|
||||
environments
|
||||
.filter((e) => e.collectActivity && e.connectionType !== 'hawser-edge')
|
||||
.map((e) => e.id)
|
||||
);
|
||||
|
||||
// Stop collectors for removed environments or those with collection disabled
|
||||
for (const envId of collectors.keys()) {
|
||||
if (!activeEnvIds.has(envId)) {
|
||||
console.log(`[EventSubprocess] Stopping collector for environment ${envId}`);
|
||||
stopEnvironmentCollector(envId);
|
||||
}
|
||||
}
|
||||
|
||||
// Start collectors for environments with collection enabled
|
||||
for (const env of environments) {
|
||||
// Skip Hawser Edge (handled by main process)
|
||||
if (env.connectionType === 'hawser-edge') continue;
|
||||
|
||||
if (env.collectActivity && !collectors.has(env.id)) {
|
||||
startEnvironmentCollector(env.id, env.name);
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('[EventSubprocess] Failed to refresh collectors:', error);
|
||||
send({ type: 'error', message: `Failed to refresh collectors: ${error}` });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle commands from main process
|
||||
*/
|
||||
function handleCommand(command: MainProcessCommand): void {
|
||||
switch (command.type) {
|
||||
case 'refresh_environments':
|
||||
console.log('[EventSubprocess] Refreshing environments...');
|
||||
refreshEventCollectors();
|
||||
break;
|
||||
|
||||
case 'shutdown':
|
||||
console.log('[EventSubprocess] Shutdown requested');
|
||||
shutdown();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Graceful shutdown
|
||||
*/
|
||||
function shutdown(): void {
|
||||
isShuttingDown = true;
|
||||
|
||||
// Stop periodic cache cleanup
|
||||
if (cacheCleanupInterval) {
|
||||
clearInterval(cacheCleanupInterval);
|
||||
cacheCleanupInterval = null;
|
||||
}
|
||||
|
||||
// Stop all environment collectors
|
||||
for (const envId of collectors.keys()) {
|
||||
stopEnvironmentCollector(envId);
|
||||
}
|
||||
|
||||
// Clear the deduplication cache
|
||||
recentEvents.clear();
|
||||
|
||||
console.log('[EventSubprocess] Stopped');
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Start the event collector
|
||||
*/
|
||||
async function start(): Promise<void> {
|
||||
console.log('[EventSubprocess] Starting container event collection...');
|
||||
|
||||
// Start collectors for all environments
|
||||
await refreshEventCollectors();
|
||||
|
||||
// Start periodic cache cleanup
|
||||
cacheCleanupInterval = setInterval(cleanupRecentEvents, CACHE_CLEANUP_INTERVAL_MS);
|
||||
console.log('[EventSubprocess] Started deduplication cache cleanup (every 30s)');
|
||||
|
||||
// Listen for commands from main process
|
||||
process.on('message', (message: MainProcessCommand) => {
|
||||
handleCommand(message);
|
||||
});
|
||||
|
||||
// Handle termination signals
|
||||
process.on('SIGTERM', shutdown);
|
||||
process.on('SIGINT', shutdown);
|
||||
|
||||
// Signal ready
|
||||
send({ type: 'ready' });
|
||||
|
||||
console.log('[EventSubprocess] Started successfully');
|
||||
}
|
||||
|
||||
// Start the subprocess
|
||||
start();
|
||||
Reference in New Issue
Block a user