correlated external traces/logs across multiple runs with external trace context

Author: ericallam

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -93,10 +93,9 @@ const { action, loader } = createActionApiRoute(
     const service = new TriggerTaskService();
 
     try {
-      const traceContext =
-        traceparent && isFromWorker /// If the request is from a worker, we should pass the trace context
-          ? { traceparent, tracestate }
-          : undefined;
+      const traceContext = isFromWorker
+        ? { traceparent, tracestate }
+        : { external: { traceparent, tracestate } };
 
       const oneTimeUseToken = await getOneTimeUseToken(authentication);
 
@@ -111,6 +110,14 @@ const { action, loader } = createActionApiRoute(
         traceContext,
       });
 
+      logger.debug("[otelContext]", {
+        taskId: params.taskId,
+        headers,
+        options: body.options,
+        isFromWorker,
+        traceContext,
+      });
+
       const idempotencyKeyExpiresAt = resolveIdempotencyKeyTTL(idempotencyKeyTTL);
 
       const result = await service.call(

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -10,8 +10,14 @@ import {
   taskRunErrorEnhancer,
   taskRunErrorToString,
   TriggerTaskRequestBody,
+  TriggerTraceContext,
 } from "@trigger.dev/core/v3";
-import { RunId, stringifyDuration } from "@trigger.dev/core/v3/isomorphic";
+import {
+  parseTraceparent,
+  RunId,
+  serializeTraceparent,
+  stringifyDuration,
+} from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClientOrTransaction } from "@trigger.dev/database";
 import { createTags } from "~/models/taskRunTag.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
@@ -253,7 +259,11 @@ export class RunEngineTriggerTaskService {
                   payload: payloadPacket.data ?? "",
                   payloadType: payloadPacket.dataType,
                   context: body.context,
-                  traceContext: event.traceContext,
+                  traceContext: this.#propagateExternalTraceContext(
+                    event.traceContext,
+                    parentRun?.traceContext,
+                    event.traceparent?.spanId
+                  ),
                   traceId: event.traceId,
                   spanId: event.spanId,
                   parentSpanId:
@@ -341,4 +351,49 @@ export class RunEngineTriggerTaskService {
       }
     });
   }
+
+  #propagateExternalTraceContext(
+    traceContext: Record<string, unknown>,
+    parentRunTraceContext: unknown,
+    parentSpanId: string | undefined
+  ): Record<string, unknown> {
+    if (!parentRunTraceContext) {
+      return traceContext;
+    }
+
+    const parsedParentRunTraceContext = TriggerTraceContext.safeParse(parentRunTraceContext);
+
+    if (!parsedParentRunTraceContext.success) {
+      return traceContext;
+    }
+
+    const { external } = parsedParentRunTraceContext.data;
+
+    if (!external) {
+      return traceContext;
+    }
+
+    if (!external.traceparent) {
+      return traceContext;
+    }
+
+    const parsedTraceparent = parseTraceparent(external.traceparent);
+
+    if (!parsedTraceparent) {
+      return traceContext;
+    }
+
+    const newExternalTraceparent = serializeTraceparent(
+      parsedTraceparent.traceId,
+      parentSpanId ?? parsedTraceparent.spanId
+    );
+
+    return {
+      ...traceContext,
+      external: {
+        ...external,
+        traceparent: newExternalTraceparent,
+      },
+    };
+  }
 }

apps/webapp/app/runEngine/types.ts

@@ -12,7 +12,7 @@ export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
   idempotencyKeyExpiresAt?: Date;
   triggerVersion?: string;
-  traceContext?: Record<string, string | undefined>;
+  traceContext?: Record<string, unknown>;
   spanParentAsLink?: boolean;
   parentAsLinkType?: "replay" | "trigger";
   batchId?: string;
@@ -119,7 +119,7 @@ export interface TriggerTaskValidator {
 export type TracedEventSpan = {
   traceId: string;
   spanId: string;
-  traceContext: Record<string, string | undefined>;
+  traceContext: Record<string, unknown>;
   traceparent?: {
     traceId: string;
     spanId: string;

apps/webapp/app/v3/eventRepository.server.ts

@@ -2,9 +2,14 @@ import { Attributes, AttributeValue, Link, trace, TraceFlags, Tracer } from "@op
 import { RandomIdGenerator } from "@opentelemetry/sdk-trace-base";
 import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
 import {
+  correctErrorStackTrace,
+  createPacketAttributesAsJson,
   ExceptionEventProperties,
   ExceptionSpanEvent,
+  flattenAttributes,
+  isExceptionSpanEvent,
   NULL_SENTINEL,
+  omit,
   PRIMARY_VARIANT,
   SemanticInternalAttributes,
   SpanEvent,
@@ -13,28 +18,24 @@ import {
   TaskEventEnvironment,
   TaskEventStyle,
   TaskRunError,
-  correctErrorStackTrace,
-  createPacketAttributesAsJson,
-  flattenAttributes,
-  isExceptionSpanEvent,
-  omit,
   unflattenAttributes,
 } from "@trigger.dev/core/v3";
+import { parseTraceparent, serializeTraceparent } from "@trigger.dev/core/v3/isomorphic";
 import { Prisma, TaskEvent, TaskEventKind, TaskEventStatus } from "@trigger.dev/database";
+import { nanoid } from "nanoid";
 import { createHash } from "node:crypto";
 import { EventEmitter } from "node:stream";
 import { Gauge } from "prom-client";
-import { $replica, PrismaClient, PrismaReplicaClient, prisma } from "~/db.server";
+import { $replica, prisma, PrismaClient, PrismaReplicaClient } from "~/db.server";
 import { env } from "~/env.server";
 import { metricsRegister } from "~/metrics.server";
+import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
 import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
 import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server";
+import { TaskEventStore, TaskEventStoreTable } from "./taskEventStore.server";
 import { startActiveSpan } from "./tracer.server";
-import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
 import { startSpan } from "./tracing.server";
-import { nanoid } from "nanoid";
-import { TaskEventStore, TaskEventStoreTable } from "./taskEventStore.server";
 
 const MAX_FLUSH_DEPTH = 5;
 
@@ -80,7 +81,7 @@ export type SetAttribute<T extends TraceAttributes> = (key: keyof T, value: T[ke
 
 export type TraceEventOptions = {
   kind?: CreatableEventKind;
-  context?: Record<string, string | undefined>;
+  context?: Record<string, unknown>;
   spanParentAsLink?: boolean;
   parentAsLinkType?: "trigger" | "replay";
   spanIdSeed?: string;
@@ -932,7 +933,7 @@ export class EventRepository {
       traceId,
       spanId,
       parentId,
-      tracestate,
+      tracestate: typeof tracestate === "string" ? tracestate : undefined,
       message: message,
       serviceName: "api server",
       serviceNamespace: "trigger.dev",
@@ -989,6 +990,11 @@ export class EventRepository {
   ): Promise<TResult> {
     const propagatedContext = extractContextFromCarrier(options.context ?? {});
 
+    logger.debug("[otelContext]", {
+      propagatedContext,
+      options,
+    });
+
     const start = process.hrtime.bigint();
     const startTime = options.startTime ?? getNowInNanoseconds();
 
@@ -1002,7 +1008,8 @@ export class EventRepository {
       : this.generateSpanId();
 
     const traceContext = {
-      traceparent: `00-${traceId}-${spanId}-01`,
+      ...options.context,
+      traceparent: serializeTraceparent(traceId, spanId),
     };
 
     const links: Link[] =
@@ -1087,7 +1094,7 @@ export class EventRepository {
       traceId,
       spanId,
       parentId,
-      tracestate,
+      tracestate: typeof tracestate === "string" ? tracestate : undefined,
       duration: options.incomplete ? 0 : duration,
       isPartial: failedWithError ? false : options.incomplete,
       isError: !!failedWithError,
@@ -1493,36 +1500,21 @@ function excludePartialEventsWithCorrespondingFullEvent(batch: CreatableEvent[])
   );
 }
 
-export function extractContextFromCarrier(carrier: Record<string, string | undefined>) {
+export function extractContextFromCarrier(carrier: Record<string, unknown>) {
   const traceparent = carrier["traceparent"];
   const tracestate = carrier["tracestate"];
 
+  if (typeof traceparent !== "string") {
+    return undefined;
+  }
+
   return {
+    ...carrier,
     traceparent: parseTraceparent(traceparent),
     tracestate,
   };
 }
 
-function parseTraceparent(traceparent?: string): { traceId: string; spanId: string } | undefined {
-  if (!traceparent) {
-    return undefined;
-  }
-
-  const parts = traceparent.split("-");
-
-  if (parts.length !== 4) {
-    return undefined;
-  }
-
-  const [version, traceId, spanId, flags] = parts;
-
-  if (version !== "00") {
-    return undefined;
-  }
-
-  return { traceId, spanId };
-}
-
 function prepareEvent(event: QueriedEvent): PreparedEvent {
   return {
     ...event,

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -19,7 +19,7 @@ export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
   idempotencyKeyExpiresAt?: Date;
   triggerVersion?: string;
-  traceContext?: Record<string, string | undefined>;
+  traceContext?: Record<string, unknown>;
   spanParentAsLink?: boolean;
   parentAsLinkType?: "replay" | "trigger";
   batchId?: string;

internal-packages/run-engine/src/engine/index.ts

@@ -385,7 +385,7 @@ export class RunEngine {
               payload,
               payloadType,
               context,
-              traceContext,
+              traceContext: traceContext as any,
               traceId,
               spanId,
               parentSpanId,

internal-packages/run-engine/src/engine/types.ts

@@ -89,7 +89,7 @@ export type TriggerParams = {
   payload: string;
   payloadType: string;
   context: any;
-  traceContext: Record<string, string | undefined>;
+  traceContext: Record<string, unknown>;
   traceId: string;
   spanId: string;
   parentSpanId?: string;

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

@@ -165,7 +165,7 @@ async function loadWorkerManifest() {
   return WorkerManifest.parse(raw);
 }
 
-async function doBootstrap() {
+async function doBootstrap(traceContext: Record<string, unknown>) {
   return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
     log("Bootstrapping worker");
 
@@ -182,6 +182,7 @@ async function doBootstrap() {
       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);
@@ -265,9 +266,9 @@ let bootstrapCache:
     }
   | undefined;
 
-async function bootstrap() {
+async function bootstrap(traceContext: Record<string, unknown>) {
   if (!bootstrapCache) {
-    bootstrapCache = await doBootstrap();
+    bootstrapCache = await doBootstrap(traceContext);
   }
 
   return bootstrapCache;
@@ -371,8 +372,9 @@ const zodIpc = new ZodIpcConnection({
       });
 
       try {
-        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } =
-          await bootstrap();
+        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } = await bootstrap(
+          traceContext
+        );
 
         _tracingSDK = tracingSDK;
 

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

@@ -156,7 +156,7 @@ async function loadWorkerManifest() {
   return WorkerManifest.parse(raw);
 }
 
-async function doBootstrap() {
+async function doBootstrap(traceContext: Record<string, unknown>) {
   return await runTimelineMetrics.measureMetric("trigger.dev/start", "bootstrap", {}, async () => {
     const workerManifest = await loadWorkerManifest();
 
@@ -173,6 +173,7 @@ async function doBootstrap() {
       forceFlushTimeoutMillis: 30_000,
       exporters: config.telemetry?.exporters ?? [],
       logExporters: config.telemetry?.logExporters ?? [],
+      externalTraceContext: traceContext.external,
     });
 
     const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
@@ -254,9 +255,9 @@ let bootstrapCache:
     }
   | undefined;
 
-async function bootstrap() {
+async function bootstrap(traceContext: Record<string, unknown>) {
   if (!bootstrapCache) {
-    bootstrapCache = await doBootstrap();
+    bootstrapCache = await doBootstrap(traceContext);
   }
 
   return bootstrapCache;
@@ -375,8 +376,9 @@ const zodIpc = new ZodIpcConnection({
       });
 
       try {
-        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } =
-          await bootstrap();
+        const { tracer, tracingSDK, consoleInterceptor, config, workerManifest } = await bootstrap(
+          traceContext
+        );
 
         _tracingSDK = tracingSDK;
 

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

@@ -617,10 +617,6 @@ export function hasOwn(obj: Object, key: string): boolean {
 function injectPropagationHeadersIfInWorker(requestInit?: RequestInit): RequestInit | undefined {
   const headers = new Headers(requestInit?.headers);
 
-  if (headers.get("x-trigger-worker") !== "true") {
-    return requestInit;
-  }
-
   const headersObject = Object.fromEntries(headers.entries());
 
   propagation.inject(context.active(), headersObject);