konnosDashboard
log-collector.ts148 lines · main
1import { AsyncLocalStorage } from 'node:async_hooks';
2
3import { sanitizeErrorMessage } from './error-sanitizer.js';
4import type { LogLine } from './isolate-runtime/types.js';
5
6/**
7 * Per-invocation log collector. Two sources feed it:
8 *
9 * 1. `ctx.log.{info,warn,error,debug}` — the structured API users should
10 * reach for first. Takes a message + optional fields object.
11 * 2. `console.{log,info,warn,error,debug}` — the fallback for anything
12 * ported from elsewhere. We patch the global `console` ONCE at module
13 * load, and route through an AsyncLocalStorage-bound collector so two
14 * concurrent invocations on the inline executor don't cross-stream.
15 *
16 * Before publish, fields whose key matches SECRET_KEY_RE are replaced with
17 * `[redacted]` so users who log their own env by accident don't leak it
18 * to the tail stream.
19 */
20
21export type LogLevel = 'debug' | 'info' | 'warn' | 'error';
22
23export interface LogEntry {
24 level: LogLevel;
25 message: string;
26 fieldsJson?: string;
27 ts: string;
28}
29
30export interface LogCollector {
31 debug(msg: string, fields?: Record<string, unknown>): void;
32 info(msg: string, fields?: Record<string, unknown>): void;
33 warn(msg: string, fields?: Record<string, unknown>): void;
34 error(msg: string, fields?: Record<string, unknown>): void;
35 drain(): LogEntry[];
36}
37
38const SECRET_KEY_RE = /^(password|secret|token|api[_-]?key|authorization)$/i;
39const MAX_ENTRIES_PER_INVOCATION = 500;
40
41const collectorAls = new AsyncLocalStorage<LogCollector>();
42
43export function createLogCollector(): LogCollector {
44 const entries: LogEntry[] = [];
45 function push(level: LogLevel, message: string, fields?: Record<string, unknown>): void {
46 if (entries.length >= MAX_ENTRIES_PER_INVOCATION) return;
47 const entry: LogEntry = {
48 level,
49 message: String(message),
50 ts: new Date().toISOString(),
51 };
52 if (fields && Object.keys(fields).length > 0) {
53 entry.fieldsJson = JSON.stringify(redact(fields));
54 }
55 entries.push(entry);
56 }
57 return {
58 debug: (m, f) => push('debug', m, f),
59 info: (m, f) => push('info', m, f),
60 warn: (m, f) => push('warn', m, f),
61 error: (m, f) => push('error', m, f),
62 drain: () => entries.splice(0, entries.length),
63 };
64}
65
66export async function runWithCollector<T>(
67 collector: LogCollector,
68 fn: () => Promise<T>,
69): Promise<T> {
70 return collectorAls.run(collector, fn);
71}
72
73/**
74 * One-time install: once called (from the module that sets up the runtime),
75 * `console.*` is redirected whenever there's an active collector in the
76 * async context. Outside an invocation the original `console` is used, so
77 * infra logs still reach stdout/stderr as normal.
78 */
79let installed = false;
80export function installConsolePatch(): void {
81 if (installed) return;
82 installed = true;
83 const orig = {
84 log: globalThis.console.log.bind(globalThis.console),
85 info: globalThis.console.info.bind(globalThis.console),
86 warn: globalThis.console.warn.bind(globalThis.console),
87 error: globalThis.console.error.bind(globalThis.console),
88 debug: globalThis.console.debug.bind(globalThis.console),
89 };
90 const route = (level: LogLevel, origFn: (...a: unknown[]) => void, args: unknown[]): void => {
91 const c = collectorAls.getStore();
92 if (!c) {
93 origFn(...args);
94 return;
95 }
96 c[level](formatArgs(args));
97 };
98 globalThis.console.log = (...args: unknown[]) => route('info', orig.log, args);
99 globalThis.console.info = (...args: unknown[]) => route('info', orig.info, args);
100 globalThis.console.warn = (...args: unknown[]) => route('warn', orig.warn, args);
101 globalThis.console.error = (...args: unknown[]) => route('error', orig.error, args);
102 globalThis.console.debug = (...args: unknown[]) => route('debug', orig.debug, args);
103}
104
105function formatArgs(args: readonly unknown[]): string {
106 return args
107 .map((a) => {
108 if (typeof a === 'string') return a;
109 if (a instanceof Error) return `${a.name}: ${a.message}`;
110 try {
111 return JSON.stringify(a);
112 } catch {
113 return String(a);
114 }
115 })
116 .join(' ');
117}
118
119function redact(fields: Record<string, unknown>): Record<string, unknown> {
120 const out: Record<string, unknown> = {};
121 for (const [k, v] of Object.entries(fields)) {
122 out[k] = SECRET_KEY_RE.test(k) ? '[redacted]' : v;
123 }
124 return out;
125}
126
127/**
128 * Ingest a stderr line emitted by a Deno isolate. Sanitizes the message
129 * (env values, tmp paths, IPs) and forwards it to a sink. Phase 1: the
130 * default sink writes to real stderr as a JSON envelope so ops can
131 * still triage panics; Task 16+ swaps in a Redis-stream publisher and
132 * wires this into the per-project log path alongside `publishInvocation`.
133 *
134 * The pool's `onLog` callback is the natural call site — runtime
135 * bootstrap (Task 14) will pass in a closure that calls this with the
136 * project's env values pre-bound.
137 */
138export function ingestIsolateLogLine(
139 line: LogLine,
140 projectId: string,
141 envValues: readonly string[] = [],
142 sink: (payload: string) => void = (s) => {
143 console.error(s);
144 },
145): void {
146 const sanitized = sanitizeErrorMessage(line.msg, envValues);
147 sink(JSON.stringify({ ...line, projectId, msg: sanitized }));
148}