Add custom telemetry exporter support (#1602)

* WIP langsmith & AI SDK integration

* Add changeset

* Add exporter support to deployed tasks

* Better support for external exporters and group exporters and instrumentations under the telemetry config property

* Missing changes

Author: ericallam

.changeset/rich-trainers-glow.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Add otel exporter support

packages/cli-v3/src/config.ts

@@ -317,8 +317,10 @@ function adaptResolveEnvVarsToSyncEnvVarsExtension(
 function getInstrumentedPackageNames(config: ResolvedConfig): Array<string> {
   const packageNames = [];
 
-  if (config.instrumentations) {
-    for (const instrumentation of config.instrumentations) {
+  if (config.instrumentations ?? config.telemetry?.instrumentations) {
+    for (const instrumentation of config.telemetry?.instrumentations ??
+      config.instrumentations ??
+      []) {
       const moduleDefinitions = (
         instrumentation as any
       ).getModuleDefinitions?.() as Array<InstrumentationModuleDefinition>;

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

@@ -148,7 +148,8 @@ async function bootstrap() {
 
   const tracingSDK = new TracingSDK({
     url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
-    instrumentations: config.instrumentations ?? [],
+    instrumentations: config.telemetry?.instrumentations ?? config.instrumentations ?? [],
+    exporters: config.telemetry?.exporters ?? [],
     diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
     forceFlushTimeoutMillis: 30_000,
   });

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

@@ -128,7 +128,8 @@ async function bootstrap() {
 
   const tracingSDK = new TracingSDK({
     url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
-    instrumentations: config.instrumentations ?? [],
+    instrumentations: config.telemetry?.instrumentations ?? config.instrumentations ?? [],
+    exporters: config.telemetry?.exporters ?? [],
     diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
     forceFlushTimeoutMillis: 30_000,
   });

packages/core/src/v3/config.ts

@@ -1,4 +1,5 @@
 import type { Instrumentation } from "@opentelemetry/instrumentation";
+import type { SpanExporter } from "@opentelemetry/sdk-trace-base";
 import type { BuildExtension } from "./build/extensions.js";
 import type { MachinePresetName } from "./schemas/common.js";
 import type { LogLevel } from "./logger/taskLogger.js";
@@ -53,9 +54,27 @@ export type TriggerConfig = {
    * Instrumentations to use for OpenTelemetry. This is useful if you want to add custom instrumentations to your tasks.
    *
    * @see https://trigger.dev/docs/config/config-file#instrumentations
+   *
+   * @deprecated Use the `telemetry.instrumentations` option instead.
    */
   instrumentations?: Array<Instrumentation>;
 
+  telemetry?: {
+    /**
+     * Instrumentations to use for OpenTelemetry. This is useful if you want to add custom instrumentations to your tasks.
+     *
+     * @see https://trigger.dev/docs/config/config-file#instrumentations
+     */
+    instrumentations?: Array<Instrumentation>;
+
+    /**
+     * Exporters to use for OpenTelemetry. This is useful if you want to add custom exporters to your tasks.
+     *
+     * @see https://trigger.dev/docs/config/config-file#exporters
+     */
+    exporters?: Array<SpanExporter>;
+  };
+
   /**
    * Specify a custom path to your tsconfig file. This is useful if you have a custom tsconfig file that you want to use.
    */

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

@@ -20,6 +20,7 @@ import {
 import {
   BatchSpanProcessor,
   NodeTracerProvider,
+  ReadableSpan,
   SimpleSpanProcessor,
   SpanExporter,
 } from "@opentelemetry/sdk-trace-node";
@@ -85,6 +86,7 @@ export type TracingSDKConfig = {
   forceFlushTimeoutMillis?: number;
   resource?: IResource;
   instrumentations?: Instrumentation[];
+  exporters?: SpanExporter[];
   diagLogLevel?: TracingDiagnosticLogLevel;
 };
 
@@ -111,6 +113,8 @@ export class TracingSDK {
       .merge(
         new Resource({
           [SemanticResourceAttributes.CLOUD_PROVIDER]: "trigger.dev",
+          [SemanticResourceAttributes.SERVICE_NAME]:
+            getEnvVar("OTEL_SERVICE_NAME") ?? "trigger.dev",
           [SemanticInternalAttributes.TRIGGER]: true,
           [SemanticInternalAttributes.CLI_VERSION]: VERSION,
         })
@@ -153,6 +157,25 @@ export class TracingSDK {
       )
     );
 
+    const externalTraceId = crypto.randomUUID();
+
+    for (const exporter of config.exporters ?? []) {
+      traceProvider.addSpanProcessor(
+        getEnvVar("OTEL_BATCH_PROCESSING_ENABLED") === "1"
+          ? new BatchSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId), {
+              maxExportBatchSize: parseInt(getEnvVar("OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"),
+              scheduledDelayMillis: parseInt(
+                getEnvVar("OTEL_SPAN_SCHEDULED_DELAY_MILLIS") ?? "200"
+              ),
+              exportTimeoutMillis: parseInt(
+                getEnvVar("OTEL_SPAN_EXPORT_TIMEOUT_MILLIS") ?? "30000"
+              ),
+              maxQueueSize: parseInt(getEnvVar("OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"),
+            })
+          : new SimpleSpanProcessor(new ExternalSpanExporterWrapper(exporter, externalTraceId))
+      );
+    }
+
     traceProvider.register();
 
     registerInstrumentations({
@@ -236,3 +259,49 @@ function setLogLevel(level: TracingDiagnosticLogLevel) {
 
   diag.setLogger(new DiagConsoleLogger(), diagLogLevel);
 }
+
+class ExternalSpanExporterWrapper {
+  constructor(
+    private underlyingExporter: SpanExporter,
+    private externalTraceId: string
+  ) {}
+
+  private transformSpan(span: ReadableSpan): ReadableSpan | undefined {
+    if (span.attributes[SemanticInternalAttributes.SPAN_PARTIAL]) {
+      // Skip partial spans
+      return;
+    }
+
+    const spanContext = span.spanContext();
+
+    return {
+      ...span,
+      spanContext: () => ({ ...spanContext, traceId: this.externalTraceId }),
+      parentSpanId: span.attributes[SemanticInternalAttributes.SPAN_ATTEMPT]
+        ? undefined
+        : span.parentSpanId,
+    };
+  }
+
+  export(spans: any[], resultCallback: (result: any) => void): void {
+    try {
+      const modifiedSpans = spans.map(this.transformSpan.bind(this));
+      this.underlyingExporter.export(
+        modifiedSpans.filter(Boolean) as ReadableSpan[],
+        resultCallback
+      );
+    } catch (e) {
+      console.error(e);
+    }
+  }
+
+  shutdown(): Promise<void> {
+    return this.underlyingExporter.shutdown();
+  }
+
+  forceFlush?(): Promise<void> {
+    return this.underlyingExporter.forceFlush
+      ? this.underlyingExporter.forceFlush()
+      : Promise.resolve();
+  }
+}

packages/core/src/v3/semanticInternalAttributes.ts

@@ -51,4 +51,5 @@ export const SemanticInternalAttributes = {
   RATE_LIMIT_LIMIT: "response.rateLimit.limit",
   RATE_LIMIT_REMAINING: "response.rateLimit.remaining",
   RATE_LIMIT_RESET: "response.rateLimit.reset",
+  SPAN_ATTEMPT: "$span.attempt",
 };

packages/core/src/v3/workers/taskExecutor.ts

@@ -231,6 +231,7 @@ export class TaskExecutor {
         kind: SpanKind.CONSUMER,
         attributes: {
           [SemanticInternalAttributes.STYLE_ICON]: "attempt",
+          [SemanticInternalAttributes.SPAN_ATTEMPT]: true,
         },
       },
       this._tracer.extractContext(traceContext),