fix(core): stop custom metric exporters breaking the metrics export

.changeset/lucky-pillows-invite.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Task metrics no longer go missing for projects that configure their own `metricExporters` or `metricReaders`, and the flush error that came with it is gone.

.github/workflows/unit-tests-packages.yml

@@ -99,6 +99,7 @@ jobs:
           pull redis:7.2
           pull testcontainers/ryuk:0.14.0
           pull electricsql/electric:1.2.4
+          pull otel/opentelemetry-collector-k8s:0.158.0@sha256:c09130a633196a5becee164411473a0932ecf223f94fda6dab5f22798ff9f376
           echo "Image pre-pull complete"
 
       - name: 📥 Download deps

internal-packages/testcontainers/package.json

@@ -22,7 +22,6 @@
     "@internal/run-ops-database": "workspace:*",
     "@testcontainers/postgresql": "^11.14.0",
     "@testcontainers/redis": "^11.14.0",
-    "@trigger.dev/core": "workspace:*",
     "std-env": "^3.9.0",
     "testcontainers": "^11.14.0",
     "tinyexec": "^0.3.0"

internal-packages/testcontainers/src/index.ts

@@ -29,6 +29,7 @@ import {
 } from "./utils";
 
 export { assertNonNullable, createPostgresContainer } from "./utils";
+export { OtelCollectorContainer, StartedOtelCollectorContainer } from "./otelCollector";
 export { laggingReplica, type LaggingModel } from "./laggingReplica";
 export { logCleanup };
 export type { MinIOConnectionConfig };

internal-packages/testcontainers/src/otelCollector.ts

@@ -0,0 +1,63 @@
+import type { StartedTestContainer } from "testcontainers";
+import { AbstractStartedContainer, GenericContainer, Wait } from "testcontainers";
+
+const OTLP_HTTP_PORT = 4318;
+const CONFIG_PATH = "/etc/otelcol-config.yaml";
+
+const CONFIG = `receivers:
+  otlp:
+    protocols:
+      http:
+        endpoint: 0.0.0.0:${OTLP_HTTP_PORT}
+exporters:
+  debug: {}
+service:
+  telemetry:
+    logs:
+      level: WARN
+  pipelines:
+    traces:
+      receivers: [otlp]
+      exporters: [debug]
+    metrics:
+      receivers: [otlp]
+      exporters: [debug]
+    logs:
+      receivers: [otlp]
+      exporters: [debug]
+`;
+
+export class OtelCollectorContainer extends GenericContainer {
+  constructor(
+    image = "otel/opentelemetry-collector-k8s:0.158.0@sha256:c09130a633196a5becee164411473a0932ecf223f94fda6dab5f22798ff9f376"
+  ) {
+    super(image);
+    this.withExposedPorts(OTLP_HTTP_PORT);
+    this.withCopyContentToContainer([{ content: CONFIG, target: CONFIG_PATH }]);
+    this.withCommand([`--config=${CONFIG_PATH}`]);
+    this.withWaitStrategy(Wait.forHttp("/v1/metrics", OTLP_HTTP_PORT).forStatusCode(405));
+    this.withStartupTimeout(120_000);
+  }
+
+  public override async start(): Promise<StartedOtelCollectorContainer> {
+    return new StartedOtelCollectorContainer(await super.start());
+  }
+}
+
+export class StartedOtelCollectorContainer extends AbstractStartedContainer {
+  constructor(startedTestContainer: StartedTestContainer) {
+    super(startedTestContainer);
+  }
+
+  public getPort(): number {
+    return super.getMappedPort(OTLP_HTTP_PORT);
+  }
+
+  /**
+   * Base URL for OTLP/HTTP, without a signal path.
+   * Example: `http://localhost:32768`
+   */
+  public getOtlpHttpUrl(): string {
+    return `http://${this.getHost()}:${this.getPort()}`;
+  }
+}

internal-packages/testcontainers/src/utils.ts

@@ -4,7 +4,6 @@ import { PostgreSqlContainer } from "@testcontainers/postgresql";
 import type { StartedRedisContainer } from "@testcontainers/redis";
 import { RedisContainer } from "@testcontainers/redis";
 import { PrismaClient } from "@trigger.dev/database";
-import { tryCatch } from "@trigger.dev/core";
 import Redis from "ioredis";
 import path from "path";
 import { isDebug } from "std-env";
@@ -16,6 +15,14 @@ import { ClickHouseContainer, runClickhouseMigrations } from "./clickhouse";
 import { MinIOContainer } from "./minio";
 import { getContainerMetadata, getTaskMetadata, logCleanup, logSetup } from "./logs";
 
+async function tryCatch<T, E = Error>(promise: Promise<T>): Promise<[E, null] | [null, T]> {
+  try {
+    return [null, await promise];
+  } catch (error) {
+    return [error as E, null];
+  }
+}
+
 /** Returns the container's connection URI with the database path swapped to `database`. */
 export function postgresUriWithDatabase(uri: string, database: string): string {
   const url = new URL(uri);

packages/core/package.json

@@ -233,6 +233,7 @@
     "@ai-sdk/provider-utils": "^1.0.22",
     "@arethetypeswrong/cli": "^0.18.5",
     "@epic-web/test-server": "^0.1.0",
+    "@internal/testcontainers": "workspace:*",
     "@trigger.dev/database": "workspace:*",
     "@types/humanize-duration": "^3.27.1",
     "@types/lodash.get": "^4.4.9",

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

@@ -0,0 +1,234 @@
+import {
+  OtelCollectorContainer,
+  type StartedOtelCollectorContainer,
+} from "@internal/testcontainers";
+
+import { metrics } from "@opentelemetry/api";
+import { ExportResultCode } from "@opentelemetry/core";
+import {
+  MetricReader,
+  type PushMetricExporter,
+  type ResourceMetrics,
+} from "@opentelemetry/sdk-metrics";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { TracingSDK } from "./tracingSDK.js";
+
+class NoopMetricExporter implements PushMetricExporter {
+  forceFlushCount = 0;
+
+  export(_metrics: ResourceMetrics, resultCallback: (result: { code: number }) => void): void {
+    resultCallback({ code: ExportResultCode.SUCCESS });
+  }
+
+  async forceFlush(): Promise<void> {
+    this.forceFlushCount++;
+  }
+
+  async shutdown(): Promise<void> {}
+}
+
+describe("TracingSDK with an external metric exporter", () => {
+  let collector: StartedOtelCollectorContainer;
+  let tracingSDK: TracingSDK;
+
+  beforeAll(async () => {
+    collector = await new OtelCollectorContainer().start();
+
+    process.env.TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS = "600000";
+
+    tracingSDK = new TracingSDK({
+      url: collector.getOtlpHttpUrl(),
+      forceFlushTimeoutMillis: 30_000,
+      diagLogLevel: "none",
+      metricExporters: [new NoopMetricExporter()],
+      hostMetrics: true,
+      hostMetricGroups: ["process.cpu", "process.memory"],
+      nodejsRuntimeMetrics: true,
+    });
+  }, 180_000);
+
+  afterAll(async () => {
+    await tracingSDK?.shutdown();
+    await collector?.stop();
+    delete process.env.TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS;
+  });
+
+  it("flushes without the collector rejecting a batch containing a NaN reading", async () => {
+    const gauge = metrics.getMeter("test").createObservableGauge("test.utilization");
+    gauge.addCallback((result) => result.observe(NaN));
+
+    await expect(tracingSDK.flush()).resolves.toBeUndefined();
+  });
+
+  it("collects from each metric reader one at a time", async () => {
+    let inFlight = 0;
+    let maxInFlight = 0;
+
+    const gauge = metrics.getMeter("test").createObservableGauge("test.concurrency");
+    gauge.addCallback(async (result) => {
+      inFlight++;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 5));
+      result.observe(1);
+      inFlight--;
+    });
+
+    await tracingSDK.flush();
+
+    expect(maxInFlight).toBe(1);
+  });
+});
+
+class FailingMetricReader extends MetricReader {
+  protected async onForceFlush(): Promise<void> {
+    throw new Error("reader flush failed");
+  }
+
+  protected async onShutdown(): Promise<void> {}
+}
+
+class FailingShutdownMetricReader extends MetricReader {
+  shutdownAttempts = 0;
+
+  protected async onForceFlush(): Promise<void> {}
+
+  protected async onShutdown(): Promise<void> {
+    this.shutdownAttempts++;
+    throw new Error(`reader shutdown failed (attempt ${this.shutdownAttempts})`);
+  }
+}
+
+class RecordingMetricReader extends MetricReader {
+  forceFlushCount = 0;
+  shutdownCount = 0;
+
+  protected async onForceFlush(): Promise<void> {
+    this.forceFlushCount++;
+  }
+
+  protected async onShutdown(): Promise<void> {
+    this.shutdownCount++;
+  }
+}
+
+function captureConsoleErrors(): { lines: string[]; restore: () => void } {
+  const lines: string[] = [];
+  const original = console.error;
+
+  console.error = (...args: unknown[]) => {
+    lines.push(args.map(String).join(" "));
+  };
+
+  return { lines, restore: () => (console.error = original) };
+}
+
+describe("TracingSDK when one metric reader fails to flush", () => {
+  let recordingReader: RecordingMetricReader;
+  let tracingSDK: TracingSDK;
+
+  beforeAll(() => {
+    recordingReader = new RecordingMetricReader();
+
+    tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingMetricReader(), recordingReader],
+    });
+  });
+
+  it("still flushes the readers after it", async () => {
+    await tracingSDK.flush().catch(() => {});
+
+    expect(recordingReader.forceFlushCount).toBeGreaterThan(0);
+  });
+
+  it("still reports the failure to the caller", async () => {
+    await expect(tracingSDK.flush()).rejects.toThrow("reader flush failed");
+  });
+
+  it("logs the failure as it happens", async () => {
+    const console = captureConsoleErrors();
+
+    await tracingSDK.flush().catch(() => {});
+    console.restore();
+
+    expect(console.lines.join("\n")).toContain("reader flush failed");
+  });
+});
+
+class OverlapRecordingMetricReader extends MetricReader {
+  static inFlight = 0;
+  static maxInFlight = 0;
+
+  protected async onForceFlush(): Promise<void> {}
+
+  protected async onShutdown(): Promise<void> {
+    OverlapRecordingMetricReader.inFlight++;
+    OverlapRecordingMetricReader.maxInFlight = Math.max(
+      OverlapRecordingMetricReader.maxInFlight,
+      OverlapRecordingMetricReader.inFlight
+    );
+    await new Promise((resolve) => setTimeout(resolve, 5));
+    OverlapRecordingMetricReader.inFlight--;
+  }
+}
+
+describe("TracingSDK shutdown", () => {
+  it("shuts down each metric reader one at a time", async () => {
+    OverlapRecordingMetricReader.inFlight = 0;
+    OverlapRecordingMetricReader.maxInFlight = 0;
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new OverlapRecordingMetricReader(), new OverlapRecordingMetricReader()],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(OverlapRecordingMetricReader.maxInFlight).toBe(1);
+  });
+
+  it("still shuts down the readers after one that fails", async () => {
+    const recordingReader = new RecordingMetricReader();
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingShutdownMetricReader(), recordingReader],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(recordingReader.shutdownCount).toBeGreaterThan(0);
+  });
+
+  it("does not retry a metric reader that failed to shut down", async () => {
+    const failingReader = new FailingShutdownMetricReader();
+
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [failingReader],
+    });
+
+    await tracingSDK.shutdown().catch(() => {});
+
+    expect(failingReader.shutdownAttempts).toBe(1);
+  });
+
+  it("reports the original shutdown failure, not a later one", async () => {
+    const tracingSDK = new TracingSDK({
+      url: "http://localhost:1",
+      forceFlushTimeoutMillis: 5_000,
+      diagLogLevel: "none",
+      metricReaders: [new FailingShutdownMetricReader()],
+    });
+
+    await expect(tracingSDK.shutdown()).rejects.toThrow("attempt 1");
+  });
+});