Add ability to "passthrough" the external trace back out into external systems

Author: ericallam

packages/cli-v3/src/entryPoints/dev-run-worker.ts

@@ -30,6 +30,7 @@ import {
   waitUntil,
   WorkerManifest,
   WorkerToExecutorMessageCatalog,
+  traceContext,
 } from "@trigger.dev/core/v3";
 import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
 import {
@@ -53,6 +54,7 @@ import {
   TracingSDK,
   usage,
   UsageTimeoutManager,
+  StandardTraceContextManager,
 } from "@trigger.dev/core/v3/workers";
 import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
 import { readFile } from "node:fs/promises";
@@ -127,6 +129,9 @@ timeout.setGlobalManager(usageTimeoutManager);
 const standardResourceCatalog = new StandardResourceCatalog();
 resourceCatalog.setGlobalResourceCatalog(standardResourceCatalog);
 
+const standardTraceContextManager = new StandardTraceContextManager();
+traceContext.setGlobalManager(standardTraceContextManager);
+
 const durableClock = new DurableClock();
 clock.setGlobalClock(durableClock);
 const runMetadataManager = new StandardMetadataManager(
@@ -165,7 +170,7 @@ async function loadWorkerManifest() {
   return WorkerManifest.parse(raw);
 }
 
-async function doBootstrap(traceContext: Record<string, unknown>) {
+async function doBootstrap() {
   return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
     log("Bootstrapping worker");
 
@@ -182,7 +187,6 @@ async function doBootstrap(traceContext: Record<string, unknown>) {
       logExporters: config.telemetry?.logExporters ?? [],
       diagLogLevel: (env.TRIGGER_OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
       forceFlushTimeoutMillis: 30_000,
-      externalTraceContext: traceContext.external,
     });
 
     const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
@@ -266,9 +270,9 @@ let bootstrapCache:
     }
   | undefined;
 
-async function bootstrap(traceContext: Record<string, unknown>) {
+async function bootstrap() {
   if (!bootstrapCache) {
-    bootstrapCache = await doBootstrap(traceContext);
+    bootstrapCache = await doBootstrap();
   }
 
   return bootstrapCache;
@@ -303,6 +307,7 @@ function resetExecutionEnvironment() {
   _sharedWorkerRuntime?.reset();
   durableClock.reset();
   taskContext.disable();
+  standardTraceContextManager.reset();
 
   log(`[${new Date().toISOString()}] Reset execution environment`);
 }
@@ -339,6 +344,7 @@ const zodIpc = new ZodIpcConnection({
 
       resetExecutionEnvironment();
 
+      standardTraceContextManager.traceContext = traceContext;
       standardRunTimelineMetricsManager.registerMetricsFromExecution(metrics, isWarmStart);
 
       if (_isRunning) {
@@ -372,9 +378,8 @@ const zodIpc = new ZodIpcConnection({
       });
 
       try {
-        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } = await bootstrap(
-          traceContext
-        );
+        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } =
+          await bootstrap();
 
         _tracingSDK = tracingSDK;
 
@@ -527,7 +532,7 @@ const zodIpc = new ZodIpcConnection({
 
           const signal = AbortSignal.any([_cancelController.signal, timeoutController.signal]);
 
-          const { result } = await executor.execute(execution, ctx, traceContext, signal);
+          const { result } = await executor.execute(execution, ctx, signal);
 
           if (_isRunning && !_isCancelled) {
             const usageSample = usage.stop(_executionMeasurement);

packages/cli-v3/src/entryPoints/managed-run-worker.ts

@@ -29,6 +29,7 @@ import {
   waitUntil,
   WorkerManifest,
   WorkerToExecutorMessageCatalog,
+  traceContext,
 } from "@trigger.dev/core/v3";
 import { TriggerTracer } from "@trigger.dev/core/v3/tracer";
 import {
@@ -53,6 +54,7 @@ import {
   TracingSDK,
   usage,
   UsageTimeoutManager,
+  StandardTraceContextManager,
 } from "@trigger.dev/core/v3/workers";
 import { ZodIpcConnection } from "@trigger.dev/core/v3/zodIpc";
 import { readFile } from "node:fs/promises";
@@ -120,6 +122,9 @@ resourceCatalog.setGlobalResourceCatalog(standardResourceCatalog);
 const durableClock = new DurableClock();
 clock.setGlobalClock(durableClock);
 
+const standardTraceContextManager = new StandardTraceContextManager();
+traceContext.setGlobalManager(standardTraceContextManager);
+
 const runMetadataManager = new StandardMetadataManager(
   apiClientManager.clientOrThrow(),
   getEnvVar("TRIGGER_STREAM_URL", getEnvVar("TRIGGER_API_URL")) ?? "https://api.trigger.dev"
@@ -156,7 +161,7 @@ async function loadWorkerManifest() {
   return WorkerManifest.parse(raw);
 }
 
-async function doBootstrap(traceContext: Record<string, unknown>) {
+async function doBootstrap() {
   return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
     const workerManifest = await loadWorkerManifest();
 
@@ -173,7 +178,6 @@ async function doBootstrap(traceContext: Record<string, unknown>) {
       forceFlushTimeoutMillis: 30_000,
       exporters: config.telemetry?.exporters ?? [],
       logExporters: config.telemetry?.logExporters ?? [],
-      externalTraceContext: traceContext.external,
     });
 
     const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
@@ -255,9 +259,9 @@ let bootstrapCache:
     }
   | undefined;
 
-async function bootstrap(traceContext: Record<string, unknown>) {
+async function bootstrap() {
   if (!bootstrapCache) {
-    bootstrapCache = await doBootstrap(traceContext);
+    bootstrapCache = await doBootstrap();
   }
 
   return bootstrapCache;
@@ -289,6 +293,7 @@ function resetExecutionEnvironment() {
   _sharedWorkerRuntime?.reset();
   durableClock.reset();
   taskContext.disable();
+  standardTraceContextManager.reset();
 
   console.log(`[${new Date().toISOString()}] Reset execution environment`);
 }
@@ -330,6 +335,8 @@ const zodIpc = new ZodIpcConnection({
 
       resetExecutionEnvironment();
 
+      standardTraceContextManager.traceContext = traceContext;
+
       const prodManager = initializeUsageManager({
         usageIntervalMs: getEnvVar("USAGE_HEARTBEAT_INTERVAL_MS"),
         usageEventUrl: getEnvVar("USAGE_EVENT_URL"),
@@ -376,9 +383,8 @@ const zodIpc = new ZodIpcConnection({
       });
 
       try {
-        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } = await bootstrap(
-          traceContext
-        );
+        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } =
+          await bootstrap();
 
         _tracingSDK = tracingSDK;
 
@@ -525,7 +531,7 @@ const zodIpc = new ZodIpcConnection({
 
           const signal = AbortSignal.any([_cancelController.signal, timeoutController.signal]);
 
-          const { result } = await executor.execute(execution, ctx, traceContext, signal);
+          const { result } = await executor.execute(execution, ctx, signal);
 
           if (_isRunning && !_isCancelled) {
             const usageSample = usage.stop(_executionMeasurement);

packages/core/src/v3/apiClient/core.ts

@@ -4,7 +4,7 @@ import { RetryOptions } from "../schemas/index.js";
 import { calculateNextRetryDelay } from "../utils/retries.js";
 import { ApiConnectionError, ApiError, ApiSchemaValidationError } from "./errors.js";
 
-import { Attributes, context, propagation, Span } from "@opentelemetry/api";
+import { Attributes, context, propagation, Span, trace } from "@opentelemetry/api";
 import { suppressTracing } from "@opentelemetry/core";
 import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
 import type { TriggerTracer } from "../tracer.js";

packages/core/src/v3/index.ts

@@ -9,6 +9,7 @@ export * from "./limits.js";
 export * from "./logger-api.js";
 export * from "./runtime-api.js";
 export * from "./task-context-api.js";
+export * from "./trace-context-api.js";
 export * from "./apiClientManager-api.js";
 export * from "./usage-api.js";
 export * from "./run-metadata-api.js";

packages/core/src/v3/otel/tracingSDK.ts

@@ -1,11 +1,6 @@
-import {
-  DiagConsoleLogger,
-  DiagLogLevel,
-  SpanContext,
-  TracerProvider,
-  diag,
-} from "@opentelemetry/api";
+import { DiagConsoleLogger, DiagLogLevel, TracerProvider, diag } from "@opentelemetry/api";
 import { logs } from "@opentelemetry/api-logs";
+import { TraceState } from "@opentelemetry/core";
 import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
 import { registerInstrumentations, type Instrumentation } from "@opentelemetry/instrumentation";
@@ -39,13 +34,13 @@ import {
   OTEL_SPAN_EVENT_COUNT_LIMIT,
 } from "../limits.js";
 import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
+import { taskContext } from "../task-context-api.js";
 import {
   TaskContextLogProcessor,
   TaskContextSpanProcessor,
 } from "../taskContext/otelProcessors.js";
+import { traceContext } from "../trace-context-api.js";
 import { getEnvVar } from "../utils/getEnv.js";
-import { taskContext } from "../task-context-api.js";
-import { TraceState } from "@opentelemetry/core";
 
 export type TracingDiagnosticLogLevel =
   | "none"
@@ -63,7 +58,6 @@ export type TracingSDKConfig = {
   exporters?: SpanExporter[];
   logExporters?: LogRecordExporter[];
   diagLogLevel?: TracingDiagnosticLogLevel;
-  externalTraceContext?: unknown;
 };
 
 const idGenerator = new RandomIdGenerator();
@@ -129,7 +123,7 @@ export class TracingSDK {
     );
 
     const externalTraceId = idGenerator.generateTraceId();
-    const externalTraceContext = extractExternalTraceContext(config.externalTraceContext);
+    const externalTraceContext = traceContext.getExternalTraceContext();
 
     for (const exporter of config.exporters ?? []) {
       spanProcessors.push(
@@ -408,30 +402,3 @@ class ExternalLogRecordExporterWrapper {
     });
   }
 }
-
-function extractExternalTraceContext(traceContext: unknown) {
-  if (typeof traceContext !== "object" || traceContext === null) {
-    return undefined;
-  }
-
-  const tracestate =
-    "tracestate" in traceContext && typeof traceContext.tracestate === "string"
-      ? traceContext.tracestate
-      : undefined;
-
-  if ("traceparent" in traceContext && typeof traceContext.traceparent === "string") {
-    const [version, traceId, spanId] = traceContext.traceparent.split("-");
-
-    if (!traceId || !spanId) {
-      return undefined;
-    }
-
-    return {
-      traceId,
-      spanId,
-      tracestate: tracestate,
-    };
-  }
-
-  return undefined;
-}

packages/core/src/v3/trace-context-api.ts

@@ -0,0 +1,5 @@
+// Split module-level variable definition into separate files to allow
+// tree-shaking on each api instance.
+import { TraceContextAPI } from "./traceContext/api.js";
+/** Entrypoint for trace context API */
+export const traceContext = TraceContextAPI.getInstance();

packages/core/src/v3/traceContext/api.ts

@@ -0,0 +1,74 @@
+import { Context } from "@opentelemetry/api";
+import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js";
+import { TraceContextManager } from "./types.js";
+
+const API_NAME = "trace-context";
+
+class NoopTraceContextManager implements TraceContextManager {
+  getTraceContext() {
+    return {};
+  }
+
+  reset() {}
+
+  getExternalTraceContext() {
+    return undefined;
+  }
+
+  extractContext(): Context {
+    throw new Error("extractContext is not implemented");
+  }
+
+  withExternalTrace<T>(fn: () => T): T {
+    return fn();
+  }
+}
+
+const NOOP_TRACE_CONTEXT_MANAGER = new NoopTraceContextManager();
+
+export class TraceContextAPI implements TraceContextManager {
+  private static _instance?: TraceContextAPI;
+
+  private constructor() {}
+
+  public static getInstance(): TraceContextAPI {
+    if (!this._instance) {
+      this._instance = new TraceContextAPI();
+    }
+
+    return this._instance;
+  }
+
+  public setGlobalManager(manager: TraceContextManager): boolean {
+    return registerGlobal(API_NAME, manager);
+  }
+
+  public disable() {
+    unregisterGlobal(API_NAME);
+  }
+
+  public reset() {
+    this.#getManager().reset();
+    this.disable();
+  }
+
+  public getTraceContext() {
+    return this.#getManager().getTraceContext();
+  }
+
+  public getExternalTraceContext() {
+    return this.#getManager().getExternalTraceContext();
+  }
+
+  public extractContext() {
+    return this.#getManager().extractContext();
+  }
+
+  public withExternalTrace<T>(fn: () => T): T {
+    return this.#getManager().withExternalTrace(fn);
+  }
+
+  #getManager(): TraceContextManager {
+    return getGlobal(API_NAME) ?? NOOP_TRACE_CONTEXT_MANAGER;
+  }
+}