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

Author: ericallam

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

@@ -148,8 +148,8 @@ async function bootstrap() {
 
   const tracingSDK = new TracingSDK({
     url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
-    instrumentations: config.instrumentations ?? [],
-    exporters: config.exporters ?? [],
+    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,8 +128,8 @@ async function bootstrap() {
 
   const tracingSDK = new TracingSDK({
     url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
-    instrumentations: config.instrumentations ?? [],
-    exporters: config.exporters ?? [],
+    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

@@ -53,10 +53,26 @@ 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>;
 
-  exporters?: Array<any>;
+  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

@@ -154,8 +154,23 @@ export class TracingSDK {
       )
     );
 
+    const externalTraceId = crypto.randomUUID();
+
     for (const exporter of config.exporters ?? []) {
-      traceProvider.addSpanProcessor(new SimpleSpanProcessor(exporter));
+      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();

references/nextjs-realtime/src/trigger/ai.ts

@@ -110,6 +110,10 @@ export const openaiStreaming = schemaTask({
     });
 
     const stream = await metadata.stream("openai", result.fullStream);
+
+    for await (const chunk of stream) {
+      logger.log("Received chunk", { chunk });
+    }
   },
 });
 

references/nextjs-realtime/trigger.config.ts

@@ -1,59 +1,13 @@
 import { defineConfig } from "@trigger.dev/sdk/v3";
 import { rscExtension } from "@trigger.dev/rsc";
 import { AISDKExporter } from "langsmith/vercel";
-import { Client } from "langsmith";
-
-class LangsmithSpanExporterWrapper {
-  constructor(
-    private underlyingExporter: any,
-    private transformSpan: (span: any) => any | undefined = (span) => {
-      if (span.attributes["$span.partial"]) {
-        // Skip partial spans
-        return;
-      }
-
-      // Check if this is an attempt span
-      if (span.name.startsWith("Attempt ")) {
-        // Create a new span that wraps the original but modifies spanContext
-        const spanContext = span.spanContext();
-
-        return {
-          ...span,
-          spanContext: () => spanContext,
-          parentSpanId: undefined,
-        };
-      }
-      return span;
-    }
-  ) {}
-
-  export(spans: any[], resultCallback: (result: any) => void): void {
-    const modifiedSpans = spans.map(this.transformSpan);
-    this.underlyingExporter.export(modifiedSpans.filter(Boolean), resultCallback);
-  }
-
-  shutdown(): Promise<void> {
-    return this.underlyingExporter.shutdown();
-  }
-
-  forceFlush?(): Promise<void> {
-    return this.underlyingExporter.forceFlush
-      ? this.underlyingExporter.forceFlush()
-      : Promise.resolve();
-  }
-}
-
-const client = new Client();
-
-const exporter = new AISDKExporter({
-  debug: true,
-  client,
-});
 
 export default defineConfig({
   project: "proj_bzhdaqhlymtuhlrcgbqy",
   dirs: ["./src/trigger"],
-  exporters: [new LangsmithSpanExporterWrapper(exporter)],
+  telemetry: {
+    exporters: [new AISDKExporter()],
+  },
   build: {
     extensions: [rscExtension({ reactDomEnvironment: "worker" })],
   },