diff --git a/compute/worker/src/server.ts b/compute/worker/src/server.ts index 038ea2d..84b6bd9 100644 --- a/compute/worker/src/server.ts +++ b/compute/worker/src/server.ts @@ -146,9 +146,16 @@ function parseBoolEnv(name: string, fallback: boolean): boolean { function buildLoggerConfig(): boolean | Record { const format = (process.env.COMPUTE_LOG_FORMAT?.trim().toLowerCase() || 'pretty'); - if (format === 'json') return true; + const level = process.env.COMPUTE_LOG_LEVEL?.trim() || 'info'; + if (format === 'json') { + return { + level, + base: null, + }; + } return { - level: process.env.COMPUTE_LOG_LEVEL?.trim() || 'info', + level, + base: null, transport: { target: 'pino-pretty', options: { @@ -705,7 +712,6 @@ async function main(): Promise { const ageMs = Date.now() - current.updatedAt; if (current.status === 'succeeded') { app.log.info({ - event: 'op.accepted', kind: req.kind, opId: current.opId, jobId: current.jobId, @@ -713,12 +719,11 @@ async function main(): Promise { action: 'reused_terminal', status: current.status, ageMs, - }); + }, 'op.accepted'); return current; } if ((current.status === 'queued' || current.status === 'running') && ageMs <= opStaleMs) { app.log.info({ - event: 'op.accepted', kind: req.kind, opId: current.opId, jobId: current.jobId, @@ -726,12 +731,11 @@ async function main(): Promise { action: 'reused_inflight', status: current.status, ageMs, - }); + }, 'op.accepted'); return current; } if ((current.status === 'queued' || current.status === 'running') && ageMs > opStaleMs) { app.log.warn({ - event: 'op.stuck_detected', kind: req.kind, opId: current.opId, jobId: current.jobId, @@ -739,7 +743,7 @@ async function main(): Promise { status: current.status, ageMs, opStaleMs, - }); + }, 'op.stuck_detected'); } } @@ -783,16 +787,14 @@ async function main(): Promise { try { await publishQueuedJob(replacement, req.payload); app.log.info({ - event: 'op.replaced', kind: req.kind, opId: replacement.opId, jobId: replacement.jobId, opKeyHash, reason: replacementReason, status: replacement.status, - }); + }, 'op.replaced'); app.log.info({ - event: 'op.accepted', kind: req.kind, opId: replacement.opId, jobId: replacement.jobId, @@ -800,7 +802,7 @@ async function main(): Promise { action: 'replaced', status: replacement.status, reason: replacementReason, - }); + }, 'op.accepted'); return replacement; } catch (error) { const failed: WorkerOperationState = { @@ -821,7 +823,6 @@ async function main(): Promise { error: failed.error, }); app.log.error({ - event: 'op.accepted', kind: req.kind, opId: failed.opId, jobId: failed.jobId, @@ -830,7 +831,7 @@ async function main(): Promise { status: failed.status, reason: replacementReason, error: failed.error?.message ?? null, - }); + }, 'op.accepted'); return failed; } } @@ -867,14 +868,13 @@ async function main(): Promise { try { await publishQueuedJob(created, req.payload); app.log.info({ - event: 'op.accepted', kind: req.kind, opId: created.opId, jobId: created.jobId, opKeyHash, action: 'created', status: created.status, - }); + }, 'op.accepted'); return created; } catch (error) { const failed: WorkerOperationState = { @@ -895,7 +895,6 @@ async function main(): Promise { error: failed.error, }); app.log.error({ - event: 'op.accepted', kind: req.kind, opId: failed.opId, jobId: failed.jobId, @@ -903,7 +902,7 @@ async function main(): Promise { action: 'created', status: failed.status, error: failed.error?.message ?? null, - }); + }, 'op.accepted'); return failed; } } @@ -943,7 +942,6 @@ async function main(): Promise { const startedAt = (request as FastifyRequest & { [REQUEST_STARTED_AT_MS_KEY]?: number })[REQUEST_STARTED_AT_MS_KEY]; const durationMs = Number.isFinite(startedAt) ? Math.max(0, Date.now() - (startedAt as number)) : -1; app.log.error({ - event: 'http.error', reqId: request.id, method: request.method, path, @@ -951,7 +949,7 @@ async function main(): Promise { durationMs, traceId: extractTraceId(request) ?? null, opId: extractOpId(request, path), - }); + }, 'http.error'); } }); @@ -1194,14 +1192,13 @@ async function main(): Promise { ...(latestProgress ? { progress: latestProgress } : {}), }); app.log.info({ - event: 'job.started', worker: input.workerLabel, kind: decoded.kind, opId: decoded.opId, jobId: decoded.jobId, queueWaitMs: queueWaitMs ?? null, deliveryCount: input.msg.info.deliveryCount, - }); + }, 'job.started'); const persistRunningState = async (updatedAt: number): Promise => { const runningOpState: WorkerOperationState = { @@ -1289,17 +1286,15 @@ async function main(): Promise { const slowJobLogThresholdMs = SLOW_JOB_LOG_THRESHOLD_MS_BY_KIND[decoded.kind]; if ((terminalDurationMs ?? 0) >= slowJobLogThresholdMs) { app.log.info({ - event: 'job.stage', worker: input.workerLabel, kind: decoded.kind, opId: decoded.opId, jobId: decoded.jobId, durationMs: terminalDurationMs ?? null, timing: resultTiming ?? null, - }); + }, 'job.stage'); } app.log.info({ - event: 'job.terminal', worker: input.workerLabel, kind: decoded.kind, opId: decoded.opId, @@ -1308,7 +1303,7 @@ async function main(): Promise { durationMs: terminalDurationMs ?? null, resultRef: extractResultRef(decoded.kind, result), timing: resultTiming ?? null, - }); + }, 'job.terminal'); } catch (error) { const message = toErrorMessage(error); const deliveryCount = input.msg.info.deliveryCount; @@ -1367,7 +1362,6 @@ async function main(): Promise { if (hasRetriesLeft) { input.msg.nak(); app.log.error({ - event: 'job.terminal', worker: input.workerLabel, kind: decoded?.kind, opId: decoded?.opId, @@ -1377,11 +1371,10 @@ async function main(): Promise { deliveryCount, maxAttempts, retryAction: 'nack_retry', - }); + }, 'job.terminal'); } else { input.msg.term(message); app.log.error({ - event: 'job.terminal', worker: input.workerLabel, kind: decoded?.kind, opId: decoded?.opId, @@ -1392,7 +1385,7 @@ async function main(): Promise { maxAttempts, retrySuppressed: isWhisperAlign ? 'whisper_align' : undefined, retryAction: 'term', - }); + }, 'job.terminal'); } } finally { if (heartbeat) clearInterval(heartbeat);