Add external log exporters and fix missing external trace exporters in deployed tasks (#2038)

* Add external log exporters and fix missing external trace exporters in deployed tasks

* Generate the external traceID correctly and exporter 3rd party logs with the external traceID as well

Author: ericallam

.changeset/cuddly-boats-press.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Add external log exporters and fix missing external trace exporters in deployed tasks

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

@@ -164,6 +164,7 @@ async function bootstrap() {
     url: env.OTEL_EXPORTER_OTLP_ENDPOINT ?? "http://0.0.0.0:4318",
     instrumentations: config.telemetry?.instrumentations ?? config.instrumentations ?? [],
     exporters: config.telemetry?.exporters ?? [],
+    logExporters: config.telemetry?.logExporters ?? [],
     diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
     forceFlushTimeoutMillis: 30_000,
   });
@@ -174,7 +175,8 @@ async function bootstrap() {
   const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
   const consoleInterceptor = new ConsoleInterceptor(
     otelLogger,
-    typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true
+    typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
+    typeof config.disableConsoleInterceptor === "boolean" ? config.disableConsoleInterceptor : false
   );
 
   const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";

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

@@ -163,6 +163,8 @@ async function bootstrap() {
     instrumentations: config.instrumentations ?? [],
     diagLogLevel: (env.OTEL_LOG_LEVEL as TracingDiagnosticLogLevel) ?? "none",
     forceFlushTimeoutMillis: 30_000,
+    exporters: config.telemetry?.exporters ?? [],
+    logExporters: config.telemetry?.logExporters ?? [],
   });
 
   const otelTracer: Tracer = tracingSDK.getTracer("trigger-dev-worker", VERSION);
@@ -171,7 +173,8 @@ async function bootstrap() {
   const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
   const consoleInterceptor = new ConsoleInterceptor(
     otelLogger,
-    typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true
+    typeof config.enableConsoleLogging === "boolean" ? config.enableConsoleLogging : true,
+    typeof config.disableConsoleInterceptor === "boolean" ? config.disableConsoleInterceptor : false
   );
 
   const configLogLevel = triggerLogLevel ?? config.logLevel ?? "info";

packages/core/src/v3/config.ts

@@ -11,6 +11,7 @@ import type {
 } from "./index.js";
 import type { LogLevel } from "./logger/taskLogger.js";
 import type { MachinePresetName } from "./schemas/common.js";
+import { LogRecordExporter } from "@opentelemetry/sdk-logs";
 
 export type CompatibilityFlag = "run_engine_v2";
 
@@ -80,6 +81,13 @@ export type TriggerConfig = {
      * @see https://trigger.dev/docs/config/config-file#exporters
      */
     exporters?: Array<SpanExporter>;
+
+    /**
+     * Log exporters to use for OpenTelemetry. This is useful if you want to add custom log exporters to your tasks.
+     *
+     * @see https://trigger.dev/docs/config/config-file#exporters
+     */
+    logExporters?: Array<LogRecordExporter>;
   };
 
   /**
@@ -131,6 +139,11 @@ export type TriggerConfig = {
    */
   enableConsoleLogging?: boolean;
 
+  /**
+   * Disable the console interceptor. This will prevent logs from being sent to the trigger.dev backend.
+   */
+  disableConsoleInterceptor?: boolean;
+
   build?: {
     /**
      * Add custom conditions to the esbuild build. For example, if you are importing `ai/rsc`, you'll need to add "react-server" condition.

packages/core/src/v3/consoleInterceptor.ts

@@ -10,12 +10,17 @@ import { clock } from "./clock-api.js";
 export class ConsoleInterceptor {
   constructor(
     private readonly logger: logsAPI.Logger,
-    private readonly sendToStdIO: boolean
+    private readonly sendToStdIO: boolean,
+    private readonly interceptingDisabled: boolean
   ) {}
 
   // Intercept the console and send logs to the OpenTelemetry logger
   // during the execution of the callback
   async intercept<T>(console: Console, callback: () => Promise<T>): Promise<T> {
+    if (this.interceptingDisabled) {
+      return await callback();
+    }
+
     // Save the original console methods
     const originalConsole = {
       log: console.log,

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

@@ -1,4 +1,5 @@
 import { DiagConsoleLogger, DiagLogLevel, TracerProvider, diag } from "@opentelemetry/api";
+import { RandomIdGenerator } from "@opentelemetry/sdk-trace-base";
 import { logs } from "@opentelemetry/api-logs";
 import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
@@ -15,6 +16,8 @@ import {
 import {
   BatchLogRecordProcessor,
   LoggerProvider,
+  LogRecordExporter,
+  ReadableLogRecord,
   SimpleLogRecordProcessor,
 } from "@opentelemetry/sdk-logs";
 import {
@@ -87,9 +90,12 @@ export type TracingSDKConfig = {
   resource?: IResource;
   instrumentations?: Instrumentation[];
   exporters?: SpanExporter[];
+  logExporters?: LogRecordExporter[];
   diagLogLevel?: TracingDiagnosticLogLevel;
 };
 
+const idGenerator = new RandomIdGenerator();
+
 export class TracingSDK {
   public readonly asyncResourceDetector = new AsyncResourceDetector();
   private readonly _logProvider: LoggerProvider;
@@ -158,7 +164,7 @@ export class TracingSDK {
       )
     );
 
-    const externalTraceId = crypto.randomUUID();
+    const externalTraceId = idGenerator.generateTraceId();
 
     for (const exporter of config.exporters ?? []) {
       traceProvider.addSpanProcessor(
@@ -210,6 +216,28 @@ export class TracingSDK {
       )
     );
 
+    for (const externalLogExporter of config.logExporters ?? []) {
+      loggerProvider.addLogRecordProcessor(
+        getEnvVar("OTEL_BATCH_PROCESSING_ENABLED") === "1"
+          ? new BatchLogRecordProcessor(
+              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId),
+              {
+                maxExportBatchSize: parseInt(getEnvVar("OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"),
+                scheduledDelayMillis: parseInt(
+                  getEnvVar("OTEL_LOG_SCHEDULED_DELAY_MILLIS") ?? "200"
+                ),
+                exportTimeoutMillis: parseInt(
+                  getEnvVar("OTEL_LOG_EXPORT_TIMEOUT_MILLIS") ?? "30000"
+                ),
+                maxQueueSize: parseInt(getEnvVar("OTEL_LOG_MAX_QUEUE_SIZE") ?? "512"),
+              }
+            )
+          : new SimpleLogRecordProcessor(
+              new ExternalLogRecordExporterWrapper(externalLogExporter, externalTraceId)
+            )
+      );
+    }
+
     this._logProvider = loggerProvider;
     this._spanExporter = spanExporter;
     this._traceProvider = traceProvider;
@@ -306,3 +334,50 @@ class ExternalSpanExporterWrapper {
       : Promise.resolve();
   }
 }
+
+class ExternalLogRecordExporterWrapper {
+  constructor(
+    private underlyingExporter: LogRecordExporter,
+    private externalTraceId: string
+  ) {}
+
+  export(logs: any[], resultCallback: (result: any) => void): void {
+    const modifiedLogs = logs.map(this.transformLogRecord.bind(this));
+
+    this.underlyingExporter.export(modifiedLogs, resultCallback);
+  }
+
+  shutdown(): Promise<void> {
+    return this.underlyingExporter.shutdown();
+  }
+
+  transformLogRecord(logRecord: ReadableLogRecord): ReadableLogRecord {
+    // If there's no spanContext, or if the externalTraceId is not set, return the original logRecord.
+    if (!logRecord.spanContext || !this.externalTraceId) {
+      return logRecord;
+    }
+
+    // Capture externalTraceId for use within the proxy's scope.
+    const { externalTraceId } = this;
+
+    return new Proxy(logRecord, {
+      get(target, prop, receiver) {
+        if (prop === "spanContext") {
+          // Intercept access to spanContext.
+          const originalSpanContext = target.spanContext;
+          // Ensure originalSpanContext exists (it should, due to the check above, but good for safety).
+          if (originalSpanContext) {
+            return {
+              ...originalSpanContext,
+              traceId: externalTraceId, // Override traceId.
+            };
+          }
+          // Fallback if, for some reason, originalSpanContext is undefined here.
+          return undefined;
+        }
+        // For all other properties, defer to the original object.
+        return Reflect.get(target, prop, receiver);
+      },
+    });
+  }
+}

pnpm-lock.yaml

@@ -27,6 +27,8 @@
     "@trigger.dev/python": "workspace:*",
     "@trigger.dev/react-hooks": "workspace:*",
     "@trigger.dev/sdk": "workspace:*",
+    "@opentelemetry/exporter-logs-otlp-http": "0.52.1",
+    "@opentelemetry/exporter-trace-otlp-http": "0.52.1",
     "@vercel/postgres": "^0.10.0",
     "ai": "4.2.5",
     "class-variance-authority": "^0.7.1",

references/d3-chat/package.json

@@ -227,6 +227,8 @@ export const interruptibleChat = schemaTask({
     prompt: z.string().describe("The prompt to chat with the AI"),
   }),
   run: async ({ prompt }, { signal }) => {
+    logger.info("interruptible-chat: starting", { prompt });
+
     const chunks: TextStreamPart<{}>[] = [];
 
     // 👇 This is a global onCancel hook, but it's inside of the run function

references/d3-chat/src/trigger/chat.ts

@@ -1,10 +1,32 @@
 import { defineConfig } from "@trigger.dev/sdk";
 import { pythonExtension } from "@trigger.dev/python/extension";
 import { installPlaywrightChromium } from "./src/extensions/playwright";
+import { OTLPLogExporter } from "@opentelemetry/exporter-logs-otlp-http";
+import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
 
 export default defineConfig({
   project: "proj_cdmymsrobxmcgjqzhdkq",
   dirs: ["./src/trigger"],
+  telemetry: {
+    logExporters: [
+      new OTLPLogExporter({
+        url: "https://api.axiom.co/v1/logs",
+        headers: {
+          Authorization: `Bearer ${process.env.AXIOM_TOKEN}`,
+          "X-Axiom-Dataset": "d3-chat-tester",
+        },
+      }),
+    ],
+    exporters: [
+      new OTLPTraceExporter({
+        url: "https://api.axiom.co/v1/traces",
+        headers: {
+          Authorization: `Bearer ${process.env.AXIOM_TOKEN}`,
+          "X-Axiom-Dataset": "d3-chat-tester",
+        },
+      }),
+    ],
+  },
   maxDuration: 3600,
   build: {
     extensions: [