From bc057054a38f1916ce43d231099cd22da2845aac Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Tue, 24 Mar 2026 14:27:41 +0000 Subject: [PATCH] feat(supervisor): add COMPUTE_TRACE_OTLP_ENDPOINT override and demote remaining logs Add optional COMPUTE_TRACE_OTLP_ENDPOINT env var to override the OTLP endpoint for supervisor-emitted spans (defaults to TRIGGER_API_URL/otel). Useful for sending spans to an OTel collector instead of the webapp. Also demotes remaining per-run logs in compute workload manager and workload server to debug/verbose. --- apps/supervisor/src/env.ts | 7 +- apps/supervisor/src/otlpPayload.ts | 63 +++++++++++++++ apps/supervisor/src/otlpTrace.test.ts | 2 +- apps/supervisor/src/otlpTrace.ts | 76 ++----------------- .../supervisor/src/workloadManager/compute.ts | 17 +++-- apps/supervisor/src/workloadServer/index.ts | 25 +++--- 6 files changed, 99 insertions(+), 91 deletions(-) create mode 100644 apps/supervisor/src/otlpPayload.ts diff --git a/apps/supervisor/src/env.ts b/apps/supervisor/src/env.ts index 74ae5d1b1..063d22930 100644 --- a/apps/supervisor/src/env.ts +++ b/apps/supervisor/src/env.ts @@ -84,6 +84,7 @@ const Env = z COMPUTE_GATEWAY_TIMEOUT_MS: z.coerce.number().int().default(30_000), COMPUTE_SNAPSHOTS_ENABLED: BoolEnv.default(false), COMPUTE_TRACE_SPANS_ENABLED: BoolEnv.default(true), + COMPUTE_TRACE_OTLP_ENDPOINT: z.string().url().optional(), // Override for span export (derived from TRIGGER_API_URL if unset) COMPUTE_SNAPSHOT_DELAY_MS: z.coerce.number().int().min(0).max(60_000).default(5_000), // Kubernetes settings @@ -168,6 +169,10 @@ const Env = z path: ["TRIGGER_WORKLOAD_API_DOMAIN"], }); } - }); + }) + .transform((data) => ({ + ...data, + COMPUTE_TRACE_OTLP_ENDPOINT: data.COMPUTE_TRACE_OTLP_ENDPOINT ?? `${data.TRIGGER_API_URL}/otel`, + })); export const env = Env.parse(stdEnv); diff --git a/apps/supervisor/src/otlpPayload.ts b/apps/supervisor/src/otlpPayload.ts new file mode 100644 index 000000000..3e5b48b53 --- /dev/null +++ b/apps/supervisor/src/otlpPayload.ts @@ -0,0 +1,63 @@ +import { randomBytes } from "crypto"; + +export interface OtlpTraceOptions { + traceId: string; + parentSpanId?: string; + spanName: string; + startTimeMs: number; + endTimeMs: number; + resourceAttributes: Record; + spanAttributes: Record; +} + +/** Build an OTLP JSON ExportTraceServiceRequest payload */ +export function buildOtlpTracePayload(opts: OtlpTraceOptions) { + const spanId = randomBytes(8).toString("hex"); + + return { + resourceSpans: [ + { + resource: { + attributes: [ + { key: "$trigger", value: { boolValue: true } }, + ...toOtlpAttributes(opts.resourceAttributes), + ], + }, + scopeSpans: [ + { + scope: { name: "supervisor.compute" }, + spans: [ + { + traceId: opts.traceId, + spanId, + parentSpanId: opts.parentSpanId, + name: opts.spanName, + kind: 3, // SPAN_KIND_CLIENT + startTimeUnixNano: String(opts.startTimeMs * 1_000_000), + endTimeUnixNano: String(opts.endTimeMs * 1_000_000), + attributes: toOtlpAttributes(opts.spanAttributes), + status: { code: 1 }, // STATUS_CODE_OK + }, + ], + }, + ], + }, + ], + }; +} + +function toOtlpAttributes( + attrs: Record +): Array<{ key: string; value: Record }> { + return Object.entries(attrs).map(([key, value]) => ({ + key, + value: toOtlpValue(value), + })); +} + +function toOtlpValue(value: string | number | boolean): Record { + if (typeof value === "string") return { stringValue: value }; + if (typeof value === "boolean") return { boolValue: value }; + if (Number.isInteger(value)) return { intValue: value }; + return { doubleValue: value }; +} diff --git a/apps/supervisor/src/otlpTrace.test.ts b/apps/supervisor/src/otlpTrace.test.ts index 765ed0282..506a4d497 100644 --- a/apps/supervisor/src/otlpTrace.test.ts +++ b/apps/supervisor/src/otlpTrace.test.ts @@ -1,5 +1,5 @@ import { describe, it, expect } from "vitest"; -import { buildOtlpTracePayload } from "./otlpTrace.js"; +import { buildOtlpTracePayload } from "./otlpPayload.js"; describe("buildOtlpTracePayload", () => { it("builds valid OTLP JSON with timing attributes", () => { diff --git a/apps/supervisor/src/otlpTrace.ts b/apps/supervisor/src/otlpTrace.ts index 7a87c056f..9cef2cb0d 100644 --- a/apps/supervisor/src/otlpTrace.ts +++ b/apps/supervisor/src/otlpTrace.ts @@ -1,83 +1,19 @@ -import { randomBytes } from "crypto"; import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger"; +import { env } from "./env.js"; +import type { buildOtlpTracePayload } from "./otlpPayload.js"; const logger = new SimpleStructuredLogger("otlp-trace"); -export interface OtlpTraceOptions { - traceId: string; - parentSpanId?: string; - spanName: string; - startTimeMs: number; - endTimeMs: number; - resourceAttributes: Record; - spanAttributes: Record; -} - -/** Build an OTLP JSON ExportTraceServiceRequest payload */ -export function buildOtlpTracePayload(opts: OtlpTraceOptions) { - const spanId = randomBytes(8).toString("hex"); - - return { - resourceSpans: [ - { - resource: { - attributes: [ - { key: "$trigger", value: { boolValue: true } }, - ...toOtlpAttributes(opts.resourceAttributes), - ], - }, - scopeSpans: [ - { - scope: { name: "supervisor.compute" }, - spans: [ - { - traceId: opts.traceId, - spanId, - parentSpanId: opts.parentSpanId, - name: opts.spanName, - kind: 3, // SPAN_KIND_CLIENT - startTimeUnixNano: String(opts.startTimeMs * 1_000_000), - endTimeUnixNano: String(opts.endTimeMs * 1_000_000), - attributes: toOtlpAttributes(opts.spanAttributes), - status: { code: 1 }, // STATUS_CODE_OK - }, - ], - }, - ], - }, - ], - }; -} - -/** Fire-and-forget: send an OTLP trace payload to the collector */ -export function sendOtlpTrace( - endpoint: string, - payload: ReturnType -) { - fetch(`${endpoint}/v1/traces`, { +/** Fire-and-forget: send an OTLP trace payload to the configured endpoint */ +export function sendOtlpTrace(payload: ReturnType) { + fetch(`${env.COMPUTE_TRACE_OTLP_ENDPOINT}/v1/traces`, { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify(payload), signal: AbortSignal.timeout(5_000), }).catch((err) => { - logger.warn("failed to send compute provision span", { + logger.warn("failed to send compute trace span", { error: err instanceof Error ? err.message : String(err), }); }); } - -function toOtlpAttributes( - attrs: Record -): Array<{ key: string; value: Record }> { - return Object.entries(attrs).map(([key, value]) => ({ - key, - value: toOtlpValue(value), - })); -} - -function toOtlpValue(value: string | number | boolean): Record { - if (typeof value === "string") return { stringValue: value }; - if (typeof value === "boolean") return { boolValue: value }; - if (Number.isInteger(value)) return { intValue: value }; - return { doubleValue: value }; -} diff --git a/apps/supervisor/src/workloadManager/compute.ts b/apps/supervisor/src/workloadManager/compute.ts index f0a126f90..892f3ecbc 100644 --- a/apps/supervisor/src/workloadManager/compute.ts +++ b/apps/supervisor/src/workloadManager/compute.ts @@ -8,7 +8,8 @@ import { } from "./types.js"; import { env } from "../env.js"; import { getRunnerId } from "../util.js"; -import { buildOtlpTracePayload, sendOtlpTrace } from "../otlpTrace.js"; +import { buildOtlpTracePayload } from "../otlpPayload.js"; +import { sendOtlpTrace } from "../otlpTrace.js"; import { tryCatch } from "@trigger.dev/core"; type ComputeWorkloadManagerOptions = WorkloadManagerOptions & { @@ -171,7 +172,7 @@ export class ComputeWorkloadManager implements WorkloadManager { } finally { event.durationMs = Math.round(performance.now() - startMs); event.ok ??= false; - this.logger.info("create instance", event); + this.logger.debug("create instance", event); } } @@ -222,7 +223,7 @@ export class ComputeWorkloadManager implements WorkloadManager { return false; } - this.logger.info("snapshot request accepted", { runnerId: opts.runnerId }); + this.logger.debug("snapshot request accepted", { runnerId: opts.runnerId }); return true; } @@ -253,7 +254,7 @@ export class ComputeWorkloadManager implements WorkloadManager { return false; } - this.logger.info("delete instance success", { runnerId }); + this.logger.debug("delete instance success", { runnerId }); return true; } @@ -312,7 +313,7 @@ export class ComputeWorkloadManager implements WorkloadManager { }); // Use the platform API URL, not the runner OTLP endpoint (which may be a VM gateway IP) - sendOtlpTrace(`${env.TRIGGER_API_URL}/otel`, payload); + sendOtlpTrace(payload); } async restore(opts: { @@ -347,7 +348,7 @@ export class ComputeWorkloadManager implements WorkloadManager { memory_mb: opts.machine.memory * 1024, }; - this.logger.debug("restore request body", { url, body }); + this.logger.verbose("restore request body", { url, body }); const startMs = performance.now(); @@ -382,7 +383,7 @@ export class ComputeWorkloadManager implements WorkloadManager { return false; } - this.logger.info("restore request success", { + this.logger.debug("restore request success", { snapshotId: opts.snapshotId, runnerId: opts.runnerId, durationMs, @@ -444,7 +445,7 @@ export class ComputeWorkloadManager implements WorkloadManager { }, }); - sendOtlpTrace(`${env.TRIGGER_API_URL}/otel`, payload); + sendOtlpTrace(payload); } } diff --git a/apps/supervisor/src/workloadServer/index.ts b/apps/supervisor/src/workloadServer/index.ts index 02c320b8f..10e85b628 100644 --- a/apps/supervisor/src/workloadServer/index.ts +++ b/apps/supervisor/src/workloadServer/index.ts @@ -27,7 +27,8 @@ import { env } from "../env.js"; import type { ComputeWorkloadManager } from "../workloadManager/compute.js"; import { TimerWheel } from "../services/timerWheel.js"; import { parseTraceparent } from "@trigger.dev/core/v3/isomorphic"; -import { buildOtlpTracePayload, sendOtlpTrace } from "../otlpTrace.js"; +import { buildOtlpTracePayload } from "../otlpPayload.js"; +import { sendOtlpTrace } from "../otlpTrace.js"; // Use the official export when upgrading to socket.io@4.8.0 interface DefaultEventsMap { @@ -469,7 +470,7 @@ export class WorkloadServer extends EventEmitter { httpServer.route("/api/v1/compute/snapshot-complete", "POST", { bodySchema: ComputeSnapshotCallbackBody, handler: async ({ reply, body }) => { - this.logger.info("Compute snapshot callback", { + this.logger.debug("Compute snapshot callback", { snapshotId: body.snapshot_id, instanceId: body.instance_id, status: body.status, @@ -504,7 +505,7 @@ export class WorkloadServer extends EventEmitter { }); if (result.success) { - this.logger.info("Suspend completion submitted", { + this.logger.debug("Suspend completion submitted", { runId, instanceId: body.instance_id, snapshotId: body.snapshot_id, @@ -553,7 +554,7 @@ export class WorkloadServer extends EventEmitter { > = io.of("/workload"); websocketServer.on("disconnect", (socket) => { - this.logger.log("[WS] disconnect", socket.id); + this.logger.verbose("[WS] disconnect", socket.id); }); websocketServer.use(async (socket, next) => { const setSocketDataFromHeader = ( @@ -635,7 +636,7 @@ export class WorkloadServer extends EventEmitter { socket.data.runFriendlyId = undefined; }; - socketLogger.log("wsServer socket connected", { ...getSocketMetadata() }); + socketLogger.debug("wsServer socket connected", { ...getSocketMetadata() }); // FIXME: where does this get set? if (socket.data.runFriendlyId) { @@ -643,7 +644,7 @@ export class WorkloadServer extends EventEmitter { } socket.on("disconnecting", (reason, description) => { - socketLogger.log("Socket disconnecting", { ...getSocketMetadata(), reason, description }); + socketLogger.verbose("Socket disconnecting", { ...getSocketMetadata(), reason, description }); if (socket.data.runFriendlyId) { runDisconnected(socket.data.runFriendlyId); @@ -651,7 +652,7 @@ export class WorkloadServer extends EventEmitter { }); socket.on("disconnect", (reason, description) => { - socketLogger.log("Socket disconnected", { ...getSocketMetadata(), reason, description }); + socketLogger.debug("Socket disconnected", { ...getSocketMetadata(), reason, description }); }); socket.on("error", (error) => { @@ -672,7 +673,7 @@ export class WorkloadServer extends EventEmitter { ...message, }); - log.log("Handling run:start"); + log.debug("Handling run:start"); try { runConnected(message.run.friendlyId); @@ -688,11 +689,13 @@ export class WorkloadServer extends EventEmitter { ...message, }); - log.log("Handling run:stop"); + log.debug("Handling run:stop"); try { runDisconnected(message.run.friendlyId); - this.runTraceContexts.delete(message.run.friendlyId); + // Don't delete trace context here - run:stop fires after each snapshot/shutdown + // but the run may be restored on a new VM and snapshot again. Trace context is + // re-populated on dequeue, and entries are small (4 strings per run). } catch (error) { log.error("run:stop error", { error }); } @@ -799,7 +802,7 @@ export class WorkloadServer extends EventEmitter { spanAttributes, }); - sendOtlpTrace(`${env.TRIGGER_API_URL}/otel`, payload); + sendOtlpTrace(payload); } registerRunTraceContext(runFriendlyId: string, ctx: RunTraceContext) {