mirror of
https://github.com/Studio-Saelix/sencho.git
synced 2026-08-31 12:48:10 +00:00
fix(logs): harden global logs with shared parsing, SSE fixes, and level filter (#539)
- Extract log parsing utilities (normalizeContainerName, parseLogTimestamp,
detectLogLevel, stripControlChars, demuxDockerLog) to shared module,
eliminating duplication between polling and SSE endpoints
- Fix timestamp regex to accept timezone offsets (+HH:MM/-HH:MM), not just Z
- Replace any[] with typed GlobalLogEntry interface
- Add SSE heartbeat (30s) to prevent reverse proxy timeouts
- Add X-Accel-Buffering: no header for nginx SSE compatibility
- Replace swallowed catch blocks with console.warn diagnostics
- Fix SSE not reconnecting on node switch (missing dep in effect array)
- Add settings change event so devMode/pollRate updates apply without remount
- Add log level filter (ALL/ERROR/WARN/INFO) to toolbar
- Replace overflow-auto div with ScrollArea for design system compliance
- Add strokeWidth={1.5} to all toolbar icons
- Include stack name in download format
- Surface fetch errors with inline banner
- Add 39 unit tests for all log parsing utilities
- Update docs with level filter documentation and refreshed screenshot
This commit is contained in:
Anso
@@ -0,0 +1,256 @@
|
||||
import { describe, it, expect } from 'vitest';
|
||||
import {
|
||||
normalizeContainerName,
|
||||
parseLogTimestamp,
|
||||
detectLogLevel,
|
||||
stripControlChars,
|
||||
demuxDockerLog,
|
||||
} from '../utils/log-parsing';
|
||||
|
||||
// ── normalizeContainerName ──────────────────────────────────────────────────
|
||||
|
||||
describe('normalizeContainerName', () => {
|
||||
it('strips dash-separated stack prefix and trailing -1 replica suffix', () => {
|
||||
expect(normalizeContainerName('mystack-redis-1', 'mystack')).toBe('redis');
|
||||
});
|
||||
|
||||
it('strips underscore-separated stack prefix and trailing _1 replica suffix', () => {
|
||||
expect(normalizeContainerName('mystack_redis_1', 'mystack')).toBe('redis');
|
||||
});
|
||||
|
||||
it('strips prefix but preserves name when no trailing replica number', () => {
|
||||
expect(normalizeContainerName('mystack-redis', 'mystack')).toBe('redis');
|
||||
});
|
||||
|
||||
it('returns raw name when stack prefix does not match', () => {
|
||||
expect(normalizeContainerName('standalone-container', 'system')).toBe('standalone-container');
|
||||
});
|
||||
|
||||
it('does not strip if prefix match is partial (no separator)', () => {
|
||||
expect(normalizeContainerName('mystackredis-1', 'mystack')).toBe('mystackredis-1');
|
||||
});
|
||||
|
||||
it('handles multi-segment service names after prefix', () => {
|
||||
expect(normalizeContainerName('mystack-my-service-1', 'mystack')).toBe('my-service');
|
||||
});
|
||||
|
||||
it('only strips the trailing -1 or _1, not higher replica numbers', () => {
|
||||
// Docker Compose v2 uses -1, v1 uses _1 for first replica.
|
||||
// Higher numbers (-2, _3) are kept since the regex only matches -1/_1.
|
||||
expect(normalizeContainerName('mystack-redis-2', 'mystack')).toBe('redis-2');
|
||||
});
|
||||
});
|
||||
|
||||
// ── parseLogTimestamp ────────────────────────────────────────────────────────
|
||||
|
||||
describe('parseLogTimestamp', () => {
|
||||
it('parses standard Z-suffix timestamp', () => {
|
||||
const result = parseLogTimestamp('2024-01-15T12:30:45.123Z some log message');
|
||||
expect(result.timestampMs).toBe(new Date('2024-01-15T12:30:45.123Z').getTime());
|
||||
expect(result.cleanMessage).toBe('some log message');
|
||||
});
|
||||
|
||||
it('parses positive timezone offset', () => {
|
||||
const result = parseLogTimestamp('2024-01-15T18:00:45.123+05:30 message with offset');
|
||||
expect(result.timestampMs).toBe(new Date('2024-01-15T18:00:45.123+05:30').getTime());
|
||||
expect(result.cleanMessage).toBe('message with offset');
|
||||
});
|
||||
|
||||
it('parses negative timezone offset', () => {
|
||||
const result = parseLogTimestamp('2024-06-01T08:00:00.000-07:00 pacific time log');
|
||||
expect(result.timestampMs).toBe(new Date('2024-06-01T08:00:00.000-07:00').getTime());
|
||||
expect(result.cleanMessage).toBe('pacific time log');
|
||||
});
|
||||
|
||||
it('parses timestamp without fractional seconds', () => {
|
||||
const result = parseLogTimestamp('2024-01-15T12:30:45Z no fractional');
|
||||
expect(result.timestampMs).toBe(new Date('2024-01-15T12:30:45Z').getTime());
|
||||
expect(result.cleanMessage).toBe('no fractional');
|
||||
});
|
||||
|
||||
it('falls back to Date.now() when no timestamp found', () => {
|
||||
const before = Date.now();
|
||||
const result = parseLogTimestamp('plain log line with no timestamp');
|
||||
const after = Date.now();
|
||||
expect(result.timestampMs).toBeGreaterThanOrEqual(before);
|
||||
expect(result.timestampMs).toBeLessThanOrEqual(after);
|
||||
expect(result.cleanMessage).toBe('plain log line with no timestamp');
|
||||
});
|
||||
|
||||
it('returns original line for empty input', () => {
|
||||
const result = parseLogTimestamp('');
|
||||
expect(result.cleanMessage).toBe('');
|
||||
});
|
||||
});
|
||||
|
||||
// ── detectLogLevel ──────────────────────────────────────────────────────────
|
||||
|
||||
describe('detectLogLevel', () => {
|
||||
// Structured key=value format
|
||||
it('detects level=error in structured logs', () => {
|
||||
expect(detectLogLevel('level=error msg="db failed"', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
it('detects level=info in structured logs', () => {
|
||||
expect(detectLogLevel('level=info msg="started"', 'STDOUT')).toBe('INFO');
|
||||
});
|
||||
|
||||
it('detects level=warn in structured logs', () => {
|
||||
expect(detectLogLevel('level=warn msg="slow query"', 'STDOUT')).toBe('WARN');
|
||||
});
|
||||
|
||||
// Bracket format
|
||||
it('detects [ERROR] bracket format', () => {
|
||||
expect(detectLogLevel('[ERROR] connection refused', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
it('detects [WARN] bracket format', () => {
|
||||
expect(detectLogLevel('[WARN] disk usage high', 'STDOUT')).toBe('WARN');
|
||||
});
|
||||
|
||||
it('detects [INFO] bracket format', () => {
|
||||
expect(detectLogLevel('[INFO] server started', 'STDERR')).toBe('INFO');
|
||||
});
|
||||
|
||||
it('detects [debug] bracket format as INFO', () => {
|
||||
expect(detectLogLevel('[debug] tracing request', 'STDOUT')).toBe('INFO');
|
||||
});
|
||||
|
||||
// Standalone keyword
|
||||
it('detects standalone ERROR keyword', () => {
|
||||
expect(detectLogLevel('ERROR: something failed', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
it('detects fatal keyword', () => {
|
||||
expect(detectLogLevel('fatal: cannot proceed', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
it('detects critical keyword', () => {
|
||||
expect(detectLogLevel('critical disk failure detected', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
it('detects panic keyword', () => {
|
||||
expect(detectLogLevel('panic: runtime error: index out of range', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
// Exception pattern
|
||||
it('detects Exception: pattern', () => {
|
||||
expect(detectLogLevel('Exception: java.lang.NullPointerException', 'STDOUT')).toBe('ERROR');
|
||||
});
|
||||
|
||||
// STDERR default
|
||||
it('defaults to ERROR for plain message from STDERR', () => {
|
||||
expect(detectLogLevel('some generic output', 'STDERR')).toBe('ERROR');
|
||||
});
|
||||
|
||||
// STDOUT default
|
||||
it('defaults to INFO for plain message from STDOUT', () => {
|
||||
expect(detectLogLevel('some generic output', 'STDOUT')).toBe('INFO');
|
||||
});
|
||||
|
||||
// Override: explicit INFO on STDERR
|
||||
it('overrides STDERR default when explicit info indicator present', () => {
|
||||
expect(detectLogLevel('level=info starting service', 'STDERR')).toBe('INFO');
|
||||
});
|
||||
|
||||
it('overrides STDERR default when [trace] bracket present', () => {
|
||||
expect(detectLogLevel('[trace] detailed operation', 'STDERR')).toBe('INFO');
|
||||
});
|
||||
|
||||
// Priority: INFO beats ERROR keyword ambiguity
|
||||
it('prioritizes INFO when both info and error-like words appear', () => {
|
||||
// "info" is checked first, so it wins even if "error" appears later.
|
||||
expect(detectLogLevel('[INFO] recovered from error state', 'STDOUT')).toBe('INFO');
|
||||
});
|
||||
});
|
||||
|
||||
// ── stripControlChars ───────────────────────────────────────────────────────
|
||||
|
||||
describe('stripControlChars', () => {
|
||||
it('strips null bytes and low control characters', () => {
|
||||
expect(stripControlChars('hello\u0000world\u0007!')).toBe('helloworld!');
|
||||
});
|
||||
|
||||
it('preserves normal text, newlines, and tabs', () => {
|
||||
// \n (0x0A) and \t (0x09) are intentionally NOT stripped
|
||||
expect(stripControlChars('line1\nline2\ttab')).toBe('line1\nline2\ttab');
|
||||
});
|
||||
|
||||
it('strips C1 control characters (0x7F-0x9F)', () => {
|
||||
expect(stripControlChars('a\u007Fb\u0080c\u009Fd')).toBe('abcd');
|
||||
});
|
||||
|
||||
it('returns empty string unchanged', () => {
|
||||
expect(stripControlChars('')).toBe('');
|
||||
});
|
||||
});
|
||||
|
||||
// ── demuxDockerLog ──────────────────────────────────────────────────────────
|
||||
|
||||
describe('demuxDockerLog', () => {
|
||||
it('parses TTY output as plain STDOUT lines', () => {
|
||||
const buf = Buffer.from('line1\nline2\n');
|
||||
const lines: Array<{ line: string; source: string }> = [];
|
||||
demuxDockerLog(buf, true, (line, source) => lines.push({ line, source }));
|
||||
expect(lines).toEqual([
|
||||
{ line: 'line1', source: 'STDOUT' },
|
||||
{ line: 'line2', source: 'STDOUT' },
|
||||
{ line: '', source: 'STDOUT' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('parses multiplexed STDOUT frame', () => {
|
||||
// Build a frame: [1, 0, 0, 0, <length BE>, ...payload]
|
||||
const payload = Buffer.from('hello stdout\n');
|
||||
const header = Buffer.alloc(8);
|
||||
header[0] = 1; // STDOUT
|
||||
header.writeUInt32BE(payload.length, 4);
|
||||
const buf = Buffer.concat([header, payload]);
|
||||
|
||||
const lines: Array<{ line: string; source: string }> = [];
|
||||
demuxDockerLog(buf, false, (line, source) => lines.push({ line, source }));
|
||||
expect(lines).toEqual([
|
||||
{ line: 'hello stdout', source: 'STDOUT' },
|
||||
{ line: '', source: 'STDOUT' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('parses multiplexed STDERR frame', () => {
|
||||
const payload = Buffer.from('error output');
|
||||
const header = Buffer.alloc(8);
|
||||
header[0] = 2; // STDERR
|
||||
header.writeUInt32BE(payload.length, 4);
|
||||
const buf = Buffer.concat([header, payload]);
|
||||
|
||||
const lines: Array<{ line: string; source: string }> = [];
|
||||
demuxDockerLog(buf, false, (line, source) => lines.push({ line, source }));
|
||||
expect(lines).toEqual([{ line: 'error output', source: 'STDERR' }]);
|
||||
});
|
||||
|
||||
it('handles multiple concatenated frames', () => {
|
||||
const p1 = Buffer.from('out1');
|
||||
const h1 = Buffer.alloc(8);
|
||||
h1[0] = 1;
|
||||
h1.writeUInt32BE(p1.length, 4);
|
||||
|
||||
const p2 = Buffer.from('err1');
|
||||
const h2 = Buffer.alloc(8);
|
||||
h2[0] = 2;
|
||||
h2.writeUInt32BE(p2.length, 4);
|
||||
|
||||
const buf = Buffer.concat([h1, p1, h2, p2]);
|
||||
const lines: Array<{ line: string; source: string }> = [];
|
||||
demuxDockerLog(buf, false, (line, source) => lines.push({ line, source }));
|
||||
expect(lines).toEqual([
|
||||
{ line: 'out1', source: 'STDOUT' },
|
||||
{ line: 'err1', source: 'STDERR' },
|
||||
]);
|
||||
});
|
||||
|
||||
it('strips control chars in TTY mode', () => {
|
||||
const buf = Buffer.from('clean\u0000text\n');
|
||||
const lines: Array<{ line: string; source: string }> = [];
|
||||
demuxDockerLog(buf, true, (line, source) => lines.push({ line, source }));
|
||||
expect(lines[0]).toEqual({ line: 'cleantext', source: 'STDOUT' });
|
||||
});
|
||||
});
|
||||
+39
-124
@@ -60,6 +60,7 @@ function invalidateNodeCaches(nodeId: number): void {
|
||||
|
||||
import { isDebugEnabled } from './utils/debug';
|
||||
import { getErrorMessage } from './utils/errors';
|
||||
import { GlobalLogEntry, normalizeContainerName, parseLogTimestamp, detectLogLevel, demuxDockerLog } from './utils/log-parsing';
|
||||
import SelfUpdateService from './services/SelfUpdateService';
|
||||
import semver from 'semver';
|
||||
import { CronExpressionParser } from 'cron-parser';
|
||||
@@ -4031,21 +4032,16 @@ app.get('/api/metrics/historical', async (req: Request, res: Response) => {
|
||||
|
||||
app.get('/api/logs/global', async (req: Request, res: Response) => {
|
||||
try {
|
||||
const debug = isDebugEnabled();
|
||||
const dockerController = DockerController.getInstance(req.nodeId);
|
||||
const containers = await dockerController.getRunningContainers();
|
||||
const allLogs: any[] = [];
|
||||
const allLogs: GlobalLogEntry[] = [];
|
||||
if (debug) console.debug('[GlobalLogs:debug] Polling snapshot starting', { containerCount: containers.length, nodeId: req.nodeId });
|
||||
|
||||
await Promise.all(containers.map(async (c) => {
|
||||
const stackName = c.Labels?.['com.docker.compose.project'] || 'system';
|
||||
const rawName = c.Names?.[0]?.replace(/^\//, '') || c.Id.substring(0, 12);
|
||||
|
||||
// Standardize naming: Strip stack name prefix if it exists
|
||||
let containerName = rawName;
|
||||
if (rawName.startsWith(`${stackName}-`)) {
|
||||
containerName = rawName.replace(`${stackName}-`, '').replace(/-1$/, '');
|
||||
} else if (rawName.startsWith(`${stackName}_`)) {
|
||||
containerName = rawName.replace(`${stackName}_`, '').replace(/_1$/, '');
|
||||
}
|
||||
const containerName = normalizeContainerName(rawName, stackName);
|
||||
|
||||
try {
|
||||
const container = dockerController.getDocker().getContainer(c.Id);
|
||||
@@ -4053,72 +4049,24 @@ app.get('/api/logs/global', async (req: Request, res: Response) => {
|
||||
const isTty = inspect.Config.Tty;
|
||||
const logsBuffer = await container.logs({ stdout: true, stderr: true, tail: 100, timestamps: true }) as Buffer;
|
||||
|
||||
const parseAndPushLog = (line: string, source: string) => {
|
||||
demuxDockerLog(logsBuffer, isTty, (line, source) => {
|
||||
if (!line.trim()) return;
|
||||
const timeMatch = line.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z)\s+(.*)/);
|
||||
let cleanMessage = line;
|
||||
let timestampMs = Date.now();
|
||||
|
||||
if (timeMatch) {
|
||||
timestampMs = new Date(timeMatch[1]).getTime();
|
||||
cleanMessage = timeMatch[2];
|
||||
}
|
||||
|
||||
// Default to INFO, or ERROR if coming from STDERR.
|
||||
let level = source === 'STDERR' ? 'ERROR' : 'INFO';
|
||||
|
||||
// 1. Explicitly check for INFO/DEBUG indicators (Overrides STDERR defaults)
|
||||
if (/level=["']?(info|debug|trace)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(info|inf|debug|dbg|trace)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(info|inf|debug|trace)(?:\s|:|\(|\[|$)/i.test(cleanMessage)) {
|
||||
level = 'INFO';
|
||||
}
|
||||
// 2. Check for WARN indicators
|
||||
else if (/level=["']?(warn|warning)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(warn|warning)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(warn|warning)(?:\s|:|\(|\[|$)/i.test(cleanMessage)) {
|
||||
level = 'WARN';
|
||||
}
|
||||
// 3. Check for ERROR indicators
|
||||
else if (/level=["']?(error|err|fatal|crit|critical|panic)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(error|err|fatal|crit|critical|panic)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(error|err|fatal|crit|critical|panic)(?:\s|:|\(|\[|$)/i.test(cleanMessage) ||
|
||||
/Exception:/i.test(cleanMessage)) {
|
||||
level = 'ERROR';
|
||||
}
|
||||
|
||||
const { timestampMs, cleanMessage } = parseLogTimestamp(line);
|
||||
const level = detectLogLevel(cleanMessage, source);
|
||||
allLogs.push({ stackName, containerName, source, level, message: cleanMessage, timestampMs });
|
||||
};
|
||||
|
||||
if (isTty) {
|
||||
// No multiplex headers. Just split by newline.
|
||||
const payload = logsBuffer.toString('utf-8').replace(/[\u0000-\u0008\u000B-\u001F\u007F-\u009F]/g, "");
|
||||
payload.split('\n').forEach(line => parseAndPushLog(line, 'STDOUT'));
|
||||
} else {
|
||||
// Parse 8-byte Docker multiplex header
|
||||
let offset = 0;
|
||||
while (offset < logsBuffer.length) {
|
||||
const streamType = logsBuffer[offset];
|
||||
const length = logsBuffer.readUInt32BE(offset + 4);
|
||||
offset += 8;
|
||||
if (offset + length > logsBuffer.length) break;
|
||||
|
||||
const payload = logsBuffer.slice(offset, offset + length).toString('utf-8');
|
||||
offset += length;
|
||||
payload.split('\n').forEach(line => parseAndPushLog(line, streamType === 2 ? 'STDERR' : 'STDOUT'));
|
||||
}
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn(`[GlobalLogs] Failed to fetch/parse logs for container ${containerName} (${c.Id.substring(0, 12)}):`, (err as Error).message);
|
||||
}
|
||||
}));
|
||||
|
||||
// Sort globally by timestamp ascending (newest bottom).
|
||||
// Limit to 500 lines - the client renders at most 300 rows at once, so
|
||||
// sending 2000 lines was wasting bandwidth and inflating JSON parse time.
|
||||
// Limit to 500 lines; the client renders at most 300 rows at once.
|
||||
allLogs.sort((a, b) => a.timestampMs - b.timestampMs);
|
||||
if (debug) console.debug('[GlobalLogs:debug] Polling snapshot complete', { totalLines: allLogs.length });
|
||||
res.json(allLogs.slice(-500));
|
||||
} catch (error) {
|
||||
console.error('[GlobalLogs] Snapshot fetch failed:', (error as Error).message);
|
||||
res.status(500).json({ error: 'Failed to fetch global logs' });
|
||||
}
|
||||
});
|
||||
@@ -4127,97 +4075,64 @@ app.get('/api/logs/global/stream', async (req: Request, res: Response) => {
|
||||
res.setHeader('Content-Type', 'text/event-stream');
|
||||
res.setHeader('Cache-Control', 'no-cache');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
// Prevent nginx from buffering SSE events (would cause burst delivery).
|
||||
res.setHeader('X-Accel-Buffering', 'no');
|
||||
res.flushHeaders();
|
||||
|
||||
const debug = isDebugEnabled();
|
||||
const dockerController = DockerController.getInstance(req.nodeId);
|
||||
const streams: NodeJS.ReadableStream[] = [];
|
||||
|
||||
// Send a heartbeat comment every 30s to keep reverse proxies from closing
|
||||
// idle connections. SSE comments (lines starting with ':') are silently
|
||||
// discarded by the browser's EventSource API.
|
||||
const heartbeat = setInterval(() => {
|
||||
if (!res.writableEnded) res.write(':heartbeat\n\n');
|
||||
}, 30_000);
|
||||
|
||||
try {
|
||||
const containers = await dockerController.getRunningContainers();
|
||||
if (debug) console.debug('[GlobalLogs:debug] SSE stream opened', { containerCount: containers.length, nodeId: req.nodeId });
|
||||
|
||||
await Promise.all(containers.map(async (c) => {
|
||||
const stackName = c.Labels?.['com.docker.compose.project'] || 'system';
|
||||
const rawName = c.Names?.[0]?.replace(/^\//, '') || c.Id.substring(0, 12);
|
||||
let containerName = rawName;
|
||||
if (rawName.startsWith(`${stackName}-`)) containerName = rawName.replace(`${stackName}-`, '').replace(/-1$/, '');
|
||||
else if (rawName.startsWith(`${stackName}_`)) containerName = rawName.replace(`${stackName}_`, '').replace(/_1$/, '');
|
||||
const containerName = normalizeContainerName(rawName, stackName);
|
||||
|
||||
try {
|
||||
const container = dockerController.getDocker().getContainer(c.Id);
|
||||
const inspect = await container.inspect();
|
||||
const isTty = inspect.Config.Tty;
|
||||
|
||||
// Dev mode gets a larger tail
|
||||
const stream = await container.logs({ follow: true, stdout: true, stderr: true, tail: 500, timestamps: true });
|
||||
streams.push(stream);
|
||||
|
||||
const processLine = (line: string, source: string) => {
|
||||
if (!line.trim()) return;
|
||||
const timeMatch = line.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d+Z)\s+(.*)/);
|
||||
let cleanMessage = line;
|
||||
let timestampMs = Date.now();
|
||||
|
||||
if (timeMatch) {
|
||||
timestampMs = new Date(timeMatch[1]).getTime();
|
||||
cleanMessage = timeMatch[2];
|
||||
}
|
||||
|
||||
// Default to INFO, or ERROR if coming from STDERR.
|
||||
let level = source === 'STDERR' ? 'ERROR' : 'INFO';
|
||||
|
||||
// 1. Explicitly check for INFO/DEBUG indicators (Overrides STDERR defaults)
|
||||
if (/level=["']?(info|debug|trace)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(info|inf|debug|dbg|trace)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(info|inf|debug|trace)(?:\s|:|\(|\[|$)/i.test(cleanMessage)) {
|
||||
level = 'INFO';
|
||||
}
|
||||
// 2. Check for WARN indicators
|
||||
else if (/level=["']?(warn|warning)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(warn|warning)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(warn|warning)(?:\s|:|\(|\[|$)/i.test(cleanMessage)) {
|
||||
level = 'WARN';
|
||||
}
|
||||
// 3. Check for ERROR indicators
|
||||
else if (/level=["']?(error|err|fatal|crit|critical|panic)["']?/i.test(cleanMessage) ||
|
||||
/\[\s*(error|err|fatal|crit|critical|panic)\s*\]/i.test(cleanMessage) ||
|
||||
/(?:\s|^)(error|err|fatal|crit|critical|panic)(?:\s|:|\(|\[|$)/i.test(cleanMessage) ||
|
||||
/Exception:/i.test(cleanMessage)) {
|
||||
level = 'ERROR';
|
||||
}
|
||||
|
||||
res.write(`data: ${JSON.stringify({ stackName, containerName, source, level, message: cleanMessage, timestampMs })}\n\n`);
|
||||
};
|
||||
|
||||
stream.on('data', (chunk: Buffer) => {
|
||||
if (isTty) {
|
||||
const payload = chunk.toString('utf-8').replace(/[\u0000-\u0008\u000B-\u001F\u007F-\u009F]/g, "");
|
||||
payload.split('\n').forEach(line => processLine(line, 'STDOUT'));
|
||||
} else {
|
||||
let offset = 0;
|
||||
while (offset < chunk.length) {
|
||||
if (offset + 8 > chunk.length) break;
|
||||
const streamType = chunk[offset];
|
||||
const length = chunk.readUInt32BE(offset + 4);
|
||||
offset += 8;
|
||||
if (offset + length > chunk.length) break;
|
||||
|
||||
const payload = chunk.slice(offset, offset + length).toString('utf-8');
|
||||
offset += length;
|
||||
payload.split('\n').forEach(line => processLine(line, streamType === 2 ? 'STDERR' : 'STDOUT'));
|
||||
demuxDockerLog(chunk, isTty, (line, source) => {
|
||||
if (!line.trim()) return;
|
||||
const { timestampMs, cleanMessage } = parseLogTimestamp(line);
|
||||
const level = detectLogLevel(cleanMessage, source);
|
||||
if (!res.writableEnded) {
|
||||
res.write(`data: ${JSON.stringify({ stackName, containerName, source, level, message: cleanMessage, timestampMs })}\n\n`);
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
} catch (err) { /* ignore */ }
|
||||
} catch (err) {
|
||||
console.warn(`[GlobalLogs] Failed to attach stream for container ${containerName} (${c.Id.substring(0, 12)}):`, (err as Error).message);
|
||||
}
|
||||
}));
|
||||
|
||||
// Cleanup when client closes the tab or switches views
|
||||
req.on('close', () => {
|
||||
clearInterval(heartbeat);
|
||||
if (debug) console.debug('[GlobalLogs:debug] SSE stream closed, cleaning up', { streamCount: streams.length });
|
||||
streams.forEach(s => {
|
||||
try { (s as any).destroy(); } catch (e) { }
|
||||
try { (s as NodeJS.ReadableStream & { destroy(): void }).destroy(); } catch { /* stream already ended */ }
|
||||
});
|
||||
});
|
||||
|
||||
} catch (error) {
|
||||
clearInterval(heartbeat);
|
||||
console.error('[GlobalLogs] SSE stream attachment failed:', (error as Error).message);
|
||||
res.write(`data: ${JSON.stringify({ level: 'ERROR', message: '[Sencho] Failed to attach global log stream.', timestampMs: Date.now(), stackName: 'system', containerName: 'backend', source: 'STDERR' })}\n\n`);
|
||||
res.end();
|
||||
}
|
||||
|
||||
@@ -872,6 +872,7 @@ class DockerController {
|
||||
res.setHeader('Content-Type', 'text/event-stream');
|
||||
res.setHeader('Cache-Control', 'no-cache');
|
||||
res.setHeader('Connection', 'keep-alive');
|
||||
res.setHeader('X-Accel-Buffering', 'no');
|
||||
res.flushHeaders();
|
||||
|
||||
try {
|
||||
|
||||
@@ -0,0 +1,140 @@
|
||||
/**
|
||||
* Shared utilities for global log parsing.
|
||||
*
|
||||
* Both the polling snapshot (`GET /api/logs/global`) and the SSE stream
|
||||
* (`GET /api/logs/global/stream`) need identical timestamp extraction,
|
||||
* log-level classification, and container-name normalization. Extracting
|
||||
* them here eliminates duplication and provides a single place to test.
|
||||
*/
|
||||
|
||||
export interface GlobalLogEntry {
|
||||
stackName: string;
|
||||
containerName: string;
|
||||
source: 'STDOUT' | 'STDERR';
|
||||
level: 'INFO' | 'WARN' | 'ERROR';
|
||||
message: string;
|
||||
timestampMs: number;
|
||||
}
|
||||
|
||||
// Matches ISO 8601 timestamps with Z or +/-HH:MM offset.
|
||||
// Docker's `timestamps: true` typically emits Z, but some logging drivers
|
||||
// and Docker configurations produce offset notation instead.
|
||||
const TIMESTAMP_RE = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2}))\s+(.*)/;
|
||||
|
||||
// Non-printable control characters that appear in TTY container logs.
|
||||
// Stripping them prevents garbled output in the UI and JSON responses.
|
||||
const CONTROL_CHARS_RE = /[\u0000-\u0008\u000B-\u001F\u007F-\u009F]/g;
|
||||
|
||||
// --- Level detection regexes (compiled once) ---
|
||||
|
||||
// Tier 1: Explicit INFO/DEBUG/TRACE indicators override the STDERR default.
|
||||
const INFO_STRUCTURED_RE = /level=["']?(info|debug|trace)["']?/i;
|
||||
const INFO_BRACKET_RE = /\[\s*(info|inf|debug|dbg|trace)\s*\]/i;
|
||||
const INFO_KEYWORD_RE = /(?:\s|^)(info|inf|debug|trace)(?:\s|:|\(|\[|$)/i;
|
||||
|
||||
// Tier 2: WARN indicators.
|
||||
const WARN_STRUCTURED_RE = /level=["']?(warn|warning)["']?/i;
|
||||
const WARN_BRACKET_RE = /\[\s*(warn|warning)\s*\]/i;
|
||||
const WARN_KEYWORD_RE = /(?:\s|^)(warn|warning)(?:\s|:|\(|\[|$)/i;
|
||||
|
||||
// Tier 3: ERROR indicators.
|
||||
const ERROR_STRUCTURED_RE = /level=["']?(error|err|fatal|crit|critical|panic)["']?/i;
|
||||
const ERROR_BRACKET_RE = /\[\s*(error|err|fatal|crit|critical|panic)\s*\]/i;
|
||||
const ERROR_KEYWORD_RE = /(?:\s|^)(error|err|fatal|crit|critical|panic)(?:\s|:|\(|\[|$)/i;
|
||||
const EXCEPTION_RE = /Exception:/i;
|
||||
|
||||
/**
|
||||
* Strip the Docker Compose stack prefix and trailing replica suffix from a
|
||||
* container name so the UI shows the clean service name.
|
||||
*
|
||||
* Examples:
|
||||
* normalizeContainerName('mystack-redis-1', 'mystack') -> 'redis'
|
||||
* normalizeContainerName('mystack_redis_1', 'mystack') -> 'redis'
|
||||
* normalizeContainerName('standalone', 'system') -> 'standalone'
|
||||
*/
|
||||
export function normalizeContainerName(rawName: string, stackName: string): string {
|
||||
if (rawName.startsWith(`${stackName}-`)) {
|
||||
return rawName.slice(stackName.length + 1).replace(/-1$/, '');
|
||||
}
|
||||
if (rawName.startsWith(`${stackName}_`)) {
|
||||
return rawName.slice(stackName.length + 1).replace(/_1$/, '');
|
||||
}
|
||||
return rawName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Extract and parse an ISO timestamp from the beginning of a Docker log line.
|
||||
* Returns the millisecond epoch value and the remainder of the line (the actual
|
||||
* log message). Falls back to `Date.now()` when no timestamp is found.
|
||||
*/
|
||||
export function parseLogTimestamp(line: string): { timestampMs: number; cleanMessage: string } {
|
||||
const match = line.match(TIMESTAMP_RE);
|
||||
if (match) {
|
||||
return {
|
||||
timestampMs: new Date(match[1]).getTime(),
|
||||
cleanMessage: match[2],
|
||||
};
|
||||
}
|
||||
return { timestampMs: Date.now(), cleanMessage: line };
|
||||
}
|
||||
|
||||
/**
|
||||
* Three-tier regex classification to detect the log level from a message.
|
||||
*
|
||||
* Priority order:
|
||||
* 1. Explicit INFO/DEBUG/TRACE keywords (overrides the STDERR default)
|
||||
* 2. WARN keywords
|
||||
* 3. ERROR/FATAL/CRIT/PANIC keywords or `Exception:` pattern
|
||||
* 4. Fallback: STDERR -> ERROR, STDOUT -> INFO
|
||||
*/
|
||||
export function detectLogLevel(message: string, source: 'STDOUT' | 'STDERR'): 'INFO' | 'WARN' | 'ERROR' {
|
||||
// Tier 1: INFO/DEBUG/TRACE (overrides STDERR default)
|
||||
if (INFO_STRUCTURED_RE.test(message) || INFO_BRACKET_RE.test(message) || INFO_KEYWORD_RE.test(message)) {
|
||||
return 'INFO';
|
||||
}
|
||||
// Tier 2: WARN
|
||||
if (WARN_STRUCTURED_RE.test(message) || WARN_BRACKET_RE.test(message) || WARN_KEYWORD_RE.test(message)) {
|
||||
return 'WARN';
|
||||
}
|
||||
// Tier 3: ERROR
|
||||
if (ERROR_STRUCTURED_RE.test(message) || ERROR_BRACKET_RE.test(message) || ERROR_KEYWORD_RE.test(message) || EXCEPTION_RE.test(message)) {
|
||||
return 'ERROR';
|
||||
}
|
||||
// Tier 4: Fallback based on stream source
|
||||
return source === 'STDERR' ? 'ERROR' : 'INFO';
|
||||
}
|
||||
|
||||
/** Strip non-printable control characters from TTY container log output. */
|
||||
export function stripControlChars(text: string): string {
|
||||
return text.replace(CONTROL_CHARS_RE, '');
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse Docker's multiplexed log stream format and call `onLine` for each
|
||||
* line. TTY containers produce raw text (no headers); non-TTY containers
|
||||
* prepend an 8-byte header per frame:
|
||||
* [streamType(1), reserved(3), payloadLength(4 BE)]
|
||||
*/
|
||||
export function demuxDockerLog(
|
||||
buf: Buffer,
|
||||
isTty: boolean,
|
||||
onLine: (line: string, source: 'STDOUT' | 'STDERR') => void,
|
||||
): void {
|
||||
if (isTty) {
|
||||
stripControlChars(buf.toString('utf-8'))
|
||||
.split('\n')
|
||||
.forEach(line => onLine(line, 'STDOUT'));
|
||||
return;
|
||||
}
|
||||
let offset = 0;
|
||||
while (offset < buf.length) {
|
||||
if (offset + 8 > buf.length) break;
|
||||
const streamType = buf[offset];
|
||||
const length = buf.readUInt32BE(offset + 4);
|
||||
offset += 8;
|
||||
if (offset + length > buf.length) break;
|
||||
const payload = buf.slice(offset, offset + length).toString('utf-8');
|
||||
offset += length;
|
||||
payload.split('\n').forEach(line => onLine(line, streamType === 2 ? 'STDERR' : 'STDOUT'));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user