github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(supervisor): add COMPUTE_TRACE_OTLP_ENDPOINT override and demote remaining logs

Browse Source 
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.
This commit is contained in:
nicktrn
2026-03-24 14:27:41 +00:00
parent f70be68220
commit bc057054a3
6 changed files with 99 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/supervisor/src/env.ts
+6 -1
View File
@@ -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);
apps/supervisor/src/otlpPayload.ts
+63
View File
@@ -0,0 +1,63 @@
import { randomBytes } from "crypto";
export interface OtlpTraceOptions {
traceId: string;
parentSpanId?: string;
spanName: string;
startTimeMs: number;
endTimeMs: number;
resourceAttributes: Record<string, string | number | boolean>;
spanAttributes: Record<string, string | number | boolean>;
}
/** 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<string, string | number | boolean>
): Array<{ key: string; value: Record<string, unknown> }> {
return Object.entries(attrs).map(([key, value]) => ({
key,
value: toOtlpValue(value),
}));
}
function toOtlpValue(value: string | number | boolean): Record<string, unknown> {
if (typeof value === "string") return { stringValue: value };
if (typeof value === "boolean") return { boolValue: value };
if (Number.isInteger(value)) return { intValue: value };
return { doubleValue: value };
}
apps/supervisor/src/otlpTrace.test.ts
+1 -1
View File
@@ -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", () => {
apps/supervisor/src/otlpTrace.ts
+6 -70
View File
@@ -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<string, string | number | boolean>;
spanAttributes: Record<string, string | number | boolean>;
}
/** 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<typeof buildOtlpTracePayload>
) {
fetch(`${endpoint}/v1/traces`, {
/** Fire-and-forget: send an OTLP trace payload to the configured endpoint */
export function sendOtlpTrace(payload: ReturnType<typeof buildOtlpTracePayload>) {
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<string, string | number | boolean>
): Array<{ key: string; value: Record<string, unknown> }> {
return Object.entries(attrs).map(([key, value]) => ({
key,
value: toOtlpValue(value),
}));
}
function toOtlpValue(value: string | number | boolean): Record<string, unknown> {
if (typeof value === "string") return { stringValue: value };
if (typeof value === "boolean") return { boolValue: value };
if (Number.isInteger(value)) return { intValue: value };
return { doubleValue: value };
}
apps/supervisor/src/workloadManager/compute.ts
+9 -8
View File
@@ -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);
}
}
apps/supervisor/src/workloadServer/index.ts
+14 -11
View File
@@ -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<WorkloadServerEvents> {
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<WorkloadServerEvents> {
});
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<WorkloadServerEvents> {
> = 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<WorkloadServerEvents> {
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<WorkloadServerEvents> {
}
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<WorkloadServerEvents> {
});
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<WorkloadServerEvents> {
...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<WorkloadServerEvents> {
...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<WorkloadServerEvents> {
spanAttributes,
});
sendOtlpTrace(`${env.TRIGGER_API_URL}/otel`, payload);
sendOtlpTrace(payload);
}
registerRunTraceContext(runFriendlyId: string, ctx: RunTraceContext) {