latest @opentelemetry packages and correlate external traces (#2334)

Author: ericallam

.changeset/five-nails-whisper.md

@@ -0,0 +1,92 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+External Trace Correlation & OpenTelemetry Package Updates.
+
+| Package | Previous Version | New Version | Change Type |
+|---------|------------------|-------------|-------------|
+| `@opentelemetry/api` | 1.9.0 | 1.9.0 | No change (stable API) |
+| `@opentelemetry/api-logs` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/core` | - | 2.0.1 | New dependency |
+| `@opentelemetry/exporter-logs-otlp-http` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/exporter-trace-otlp-http` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/instrumentation` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/instrumentation-fetch` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/resources` | 1.25.1 | 2.0.1 | Major update |
+| `@opentelemetry/sdk-logs` | 0.52.1 | 0.203.0 | Major update |
+| `@opentelemetry/sdk-node` | 0.52.1 | - | Removed (functionality consolidated) |
+| `@opentelemetry/sdk-trace-base` | 1.25.1 | 2.0.1 | Major update |
+| `@opentelemetry/sdk-trace-node` | 1.25.1 | 2.0.1 | Major update |
+| `@opentelemetry/semantic-conventions` | 1.25.1 | 1.36.0 | Minor update |
+
+### External trace correlation and propagation
+
+We will now correlate your external traces with trigger.dev traces and logs when using our external exporters:
+
+```ts
+import { defineConfig } from "@trigger.dev/sdk";
+import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
+
+export default defineConfig({
+  project: process.env.TRIGGER_PROJECT_REF,
+  dirs: ["./src/trigger"],
+  telemetry: {
+    logExporters: [
+      new OTLPLogExporter({
+        url: "https://api.axiom.co/v1/logs",
+        headers: {
+          Authorization: `Bearer ${process.env.AXIOM_TOKEN}`,
+          "X-Axiom-Dataset": "test",
+        },
+      }),
+    ],
+    exporters: [
+      new OTLPTraceExporter({
+        url: "https://api.axiom.co/v1/traces",
+        headers: {
+          Authorization: `Bearer ${process.env.AXIOM_TOKEN}`,
+          "X-Axiom-Dataset": "test",
+        },
+      }),
+    ],
+  },
+  maxDuration: 3600,
+});
+```
+
+You can also now propagate your external trace context when calling back into your own backend infra from inside a trigger.dev task:
+
+```ts
+import { otel, task } from "@trigger.dev/sdk";
+import { context, propagation } from "@opentelemetry/api";
+
+async function callNextjsApp() {
+  return await otel.withExternalTrace(async () => {
+    const headersObject = {};
+
+    // Now context.active() refers to your external trace context
+    propagation.inject(context.active(), headersObject);
+
+    const result = await fetch("http://localhost:3000/api/demo-call-from-trigger", {
+      headers: new Headers(headersObject),
+      method: "POST",
+      body: JSON.stringify({
+        message: "Hello from Trigger.dev",
+      }),
+    });
+
+    return result.json();
+  });
+}
+
+export const myTask = task({
+  id: "my-task",
+  run: async (payload: any) => {
+    await callNextjsApp()
+  }
+})
+```
+
+

apps/webapp/app/presenters/v3/SpanPresenter.server.ts

@@ -4,9 +4,10 @@ import {
   SemanticInternalAttributes,
   TaskRunContext,
   TaskRunError,
+  TriggerTraceContext,
   V3TaskRunContext,
 } from "@trigger.dev/core/v3";
-import { AttemptId, getMaxDuration } from "@trigger.dev/core/v3/isomorphic";
+import { AttemptId, getMaxDuration, parseTraceparent } from "@trigger.dev/core/v3/isomorphic";
 import { RUNNING_STATUSES } from "~/components/runs/v3/TaskRunStatus";
 import { logger } from "~/services/logger.server";
 import { eventRepository, rehydrateAttribute } from "~/v3/eventRepository.server";
@@ -173,6 +174,8 @@ export class SpanPresenter extends BasePresenter {
 
     const context = await this.#getTaskRunContext({ run, machine: machine ?? undefined });
 
+    const externalTraceId = this.#getExternalTraceId(run.traceContext);
+
     return {
       id: run.id,
       friendlyId: run.friendlyId,
@@ -234,6 +237,7 @@ export class SpanPresenter extends BasePresenter {
       spanId: run.spanId,
       isCached: !!span.originalRun,
       machinePreset: machine?.name,
+      externalTraceId,
     };
   }
 
@@ -272,6 +276,7 @@ export class SpanPresenter extends BasePresenter {
         id: true,
         spanId: true,
         traceId: true,
+        traceContext: true,
         //metadata
         number: true,
         taskIdentifier: true,
@@ -574,4 +579,26 @@ export class SpanPresenter extends BasePresenter {
   async #getV4TaskRunContext({ run }: { run: FindRunResult }): Promise<TaskRunContext> {
     return engine.resolveTaskRunContext(run.id);
   }
+
+  #getExternalTraceId(traceContext: unknown) {
+    if (!traceContext) {
+      return;
+    }
+
+    const parsedTraceContext = TriggerTraceContext.safeParse(traceContext);
+
+    if (!parsedTraceContext.success) {
+      return;
+    }
+
+    const externalTraceparent = parsedTraceContext.data.external?.traceparent;
+
+    if (!externalTraceparent) {
+      return;
+    }
+
+    const parsedTraceparent = parseTraceparent(externalTraceparent);
+
+    return parsedTraceparent?.traceId;
+  }
 }

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/routes/api.v2.tasks.batch.ts

@@ -103,10 +103,9 @@ const { action, loader } = createActionApiRoute(
       return cachedResponse;
     }
 
-    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 service = new RunEngineBatchTriggerService(batchProcessingStrategy ?? undefined);
 

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -743,6 +743,12 @@ function RunBody({
                   <Property.Label>Run Engine</Property.Label>
                   <Property.Value>{run.engine}</Property.Value>
                 </Property.Item>
+                {run.externalTraceId && (
+                  <Property.Item>
+                    <Property.Label>External Trace ID</Property.Label>
+                    <Property.Value>{run.externalTraceId}</Property.Value>
+                  </Property.Item>
+                )}
                 {isAdmin && (
                   <div className="border-t border-yellow-500/50 pt-2">
                     <Paragraph spacing variant="small" className="text-yellow-500">

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

@@ -42,7 +42,7 @@ export type BatchProcessingOptions = z.infer<typeof BatchProcessingOptions>;
 
 export type BatchTriggerTaskServiceOptions = {
   triggerVersion?: string;
-  traceContext?: Record<string, string | undefined>;
+  traceContext?: Record<string, string | undefined | Record<string, string | undefined>>;
   spanParentAsLink?: boolean;
   oneTimeUseToken?: string;
 };

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
+  ): TriggerTraceContext {
+    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/environmentVariableRules.server.ts

@@ -8,8 +8,6 @@ type VariableRule =
 const blacklistedVariables: VariableRule[] = [
   { type: "exact", key: "TRIGGER_SECRET_KEY" },
   { type: "exact", key: "TRIGGER_API_URL" },
-  { type: "prefix", prefix: "OTEL_" },
-  { type: "whitelist", key: "OTEL_LOG_LEVEL" },
 ];
 
 export function removeBlacklistedVariables(