From fa7eea39d8c07de8466d1e714bb87531b2ecbc4d Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Fri, 14 Aug 2026 08:40:07 +0100 Subject: [PATCH] fix(core): stop custom metric exporters breaking the metrics export (#4613) ## Summary Projects that configure their own `metricExporters` or `metricReaders` in `trigger.config.ts` were losing task metrics on nearly every run, and seeing an unexplained `Failed to flush tracingSDK` alongside `OTLPExporterError: Bad Request` in their run logs. Spans and logs kept working, so the runs otherwise looked healthy. ## Root cause and fix Every configured exporter gets its own `PeriodicExportingMetricReader`, and `meterProvider.forceFlush()` fans out across all readers with `Promise.all`, so two collections can land on the same millisecond. `@opentelemetry/host-metrics` divides by the elapsed interval to compute `process.cpu.utilization` ([common.ts](https://github.com/open-telemetry/opentelemetry-js-contrib/blob/main/packages/host-metrics/src/stats/common.ts)), so a zero interval yields `0/0`. `JSON.stringify(NaN)` is `null`, and a collector rejects `"asDouble": null` with a 400 that drops the **entire** request, not just the offending point. `flush()` and `shutdown()` now walk the metric readers one at a time, so collections can no longer share a timestamp. Each reader is isolated, so one failing reader cannot skip the readers behind it, and every failure is logged with the reader that produced it. The first error is still rethrown, so callers see failures exactly as before. As a second layer, non-finite data points are dropped just before our own export, so a metric that divides by zero cannot take the rest of the batch with it. Exporters and readers supplied through `trigger.config.ts` are untouched by that filter and still receive raw data. The trade-off is that configured exporters now flush after the built-in one rather than alongside it, so flush latency is the sum rather than the max. An internal test package's dependency on core was replaced with a local helper, because core now needs that package in `devDependencies` and the two together formed a workspace cycle. ## Verification Tested against a real collector in a container: a batch containing a `NaN` reading is rejected with a 400 without the fix and accepted with it, and a single flush is asserted to collect from one reader at a time. --- .changeset/lucky-pillows-invite.md | 5 + .github/workflows/unit-tests-packages.yml | 1 + internal-packages/testcontainers/package.json | 1 - internal-packages/testcontainers/src/index.ts | 1 + .../testcontainers/src/otelCollector.ts | 63 +++++ internal-packages/testcontainers/src/utils.ts | 9 +- packages/core/package.json | 1 + packages/core/src/v3/otel/tracingSDK.test.ts | 234 ++++++++++++++++++ packages/core/src/v3/otel/tracingSDK.ts | 35 ++- .../src/v3/otel/tracingSDKShutdown.test.ts | 19 ++ .../src/v3/taskContext/otelProcessors.test.ts | 144 +++++++++++ .../core/src/v3/taskContext/otelProcessors.ts | 33 ++- pnpm-lock.yaml | 6 +- 13 files changed, 544 insertions(+), 8 deletions(-) create mode 100644 .changeset/lucky-pillows-invite.md create mode 100644 internal-packages/testcontainers/src/otelCollector.ts create mode 100644 packages/core/src/v3/otel/tracingSDK.test.ts create mode 100644 packages/core/src/v3/otel/tracingSDKShutdown.test.ts create mode 100644 packages/core/src/v3/taskContext/otelProcessors.test.ts diff --git a/.changeset/lucky-pillows-invite.md b/.changeset/lucky-pillows-invite.md new file mode 100644 index 000000000..2f43f51ff --- /dev/null +++ b/.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. diff --git a/.github/workflows/unit-tests-packages.yml b/.github/workflows/unit-tests-packages.yml index 465174fb9..ceb0b7b49 100644 --- a/.github/workflows/unit-tests-packages.yml +++ b/.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 diff --git a/internal-packages/testcontainers/package.json b/internal-packages/testcontainers/package.json index 291c1314c..df0df141c 100644 --- a/internal-packages/testcontainers/package.json +++ b/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" diff --git a/internal-packages/testcontainers/src/index.ts b/internal-packages/testcontainers/src/index.ts index e1cd3d25a..da8c27980 100644 --- a/internal-packages/testcontainers/src/index.ts +++ b/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 }; diff --git a/internal-packages/testcontainers/src/otelCollector.ts b/internal-packages/testcontainers/src/otelCollector.ts new file mode 100644 index 000000000..7df8db12e --- /dev/null +++ b/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 { + 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()}`; + } +} diff --git a/internal-packages/testcontainers/src/utils.ts b/internal-packages/testcontainers/src/utils.ts index 385b4113f..a523eb589 100644 --- a/internal-packages/testcontainers/src/utils.ts +++ b/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(promise: Promise): 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); diff --git a/packages/core/package.json b/packages/core/package.json index e2ee6f064..75c95cd8f 100644 --- a/packages/core/package.json +++ b/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", diff --git a/packages/core/src/v3/otel/tracingSDK.test.ts b/packages/core/src/v3/otel/tracingSDK.test.ts new file mode 100644 index 000000000..3595fceff --- /dev/null +++ b/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 { + this.forceFlushCount++; + } + + async shutdown(): Promise {} +} + +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 { + throw new Error("reader flush failed"); + } + + protected async onShutdown(): Promise {} +} + +class FailingShutdownMetricReader extends MetricReader { + shutdownAttempts = 0; + + protected async onForceFlush(): Promise {} + + protected async onShutdown(): Promise { + this.shutdownAttempts++; + throw new Error(`reader shutdown failed (attempt ${this.shutdownAttempts})`); + } +} + +class RecordingMetricReader extends MetricReader { + forceFlushCount = 0; + shutdownCount = 0; + + protected async onForceFlush(): Promise { + this.forceFlushCount++; + } + + protected async onShutdown(): Promise { + 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 {} + + protected async onShutdown(): Promise { + 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"); + }); +}); diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 0b3a66a87..0f4ea82a2 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -100,6 +100,7 @@ export class TracingSDK { private readonly _spanExporter: SpanExporter; private readonly _traceProvider: NodeTracerProvider; private readonly _meterProvider: MeterProvider; + private readonly _metricReaders: MetricReader[]; public readonly getLogger: LoggerProvider["getLogger"]; public readonly getTracer: TracerProvider["getTracer"]; @@ -318,6 +319,7 @@ export class TracingSDK { }); this._meterProvider = meterProvider; + this._metricReaders = metricReaders; metrics.setGlobalMeterProvider(meterProvider); if (config.hostMetrics) { @@ -348,15 +350,44 @@ export class TracingSDK { await Promise.all([ this._traceProvider.forceFlush(), this._logProvider.forceFlush(), - this._meterProvider.forceFlush(), + this._flushMetricReadersSerially(), ]); } + private async _flushMetricReadersSerially() { + await this._eachMetricReaderSerially("flush", (reader) => reader.forceFlush()); + } + + private async _shutdownMetricReadersSerially() { + await this._eachMetricReaderSerially("shut down", (reader) => reader.shutdown()); + await this._meterProvider.shutdown(); + } + + private async _eachMetricReaderSerially( + action: string, + run: (reader: MetricReader) => Promise + ) { + const errors: unknown[] = []; + + for (const reader of this._metricReaders) { + try { + await run(reader); + } catch (error) { + console.error(`Failed to ${action} metric reader ${reader.constructor.name}`, error); + errors.push(error); + } + } + + if (errors.length > 0) { + throw errors[0]; + } + } + public async shutdown() { await Promise.all([ this._traceProvider.shutdown(), this._logProvider.shutdown(), - this._meterProvider.shutdown(), + this._shutdownMetricReadersSerially(), ]); } } diff --git a/packages/core/src/v3/otel/tracingSDKShutdown.test.ts b/packages/core/src/v3/otel/tracingSDKShutdown.test.ts new file mode 100644 index 000000000..6d01eb72c --- /dev/null +++ b/packages/core/src/v3/otel/tracingSDKShutdown.test.ts @@ -0,0 +1,19 @@ +import { metrics } from "@opentelemetry/api"; +import { describe, expect, it } from "vitest"; +import { TracingSDK } from "./tracingSDK.js"; + +describe("TracingSDK shutdown", () => { + it("leaves the meter provider shut down", async () => { + const tracingSDK = new TracingSDK({ + url: "http://localhost:1", + forceFlushTimeoutMillis: 5_000, + diagLogLevel: "none", + }); + + const meterBeforeShutdown = metrics.getMeter("test"); + + await tracingSDK.shutdown().catch(() => {}); + + expect(metrics.getMeter("test")).not.toBe(meterBeforeShutdown); + }); +}); diff --git a/packages/core/src/v3/taskContext/otelProcessors.test.ts b/packages/core/src/v3/taskContext/otelProcessors.test.ts new file mode 100644 index 000000000..5bd1cd4e4 --- /dev/null +++ b/packages/core/src/v3/taskContext/otelProcessors.test.ts @@ -0,0 +1,144 @@ +import { + OtelCollectorContainer, + type StartedOtelCollectorContainer, +} from "@internal/testcontainers"; +import { ExportResultCode } from "@opentelemetry/core"; +import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http"; +import { + AggregationTemporality, + DataPointType, + InstrumentType, + type MetricData, + type PushMetricExporter, + type ResourceMetrics, +} from "@opentelemetry/sdk-metrics"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { BufferingMetricExporter } from "./otelProcessors.js"; + +class RecordingMetricExporter implements PushMetricExporter { + exported: ResourceMetrics[] = []; + + export(metrics: ResourceMetrics, resultCallback: (result: { code: number }) => void): void { + this.exported.push(metrics); + resultCallback({ code: ExportResultCode.SUCCESS }); + } + + async forceFlush(): Promise {} + async shutdown(): Promise {} +} + +function gauge(name: string, values: number[]): MetricData { + return { + descriptor: { + name, + description: "", + unit: "1", + type: InstrumentType.OBSERVABLE_GAUGE, + valueType: 1, + }, + aggregationTemporality: AggregationTemporality.CUMULATIVE, + dataPointType: DataPointType.GAUGE, + dataPoints: values.map((value) => ({ + attributes: { "process.cpu.state": "user" }, + startTime: [1786481102, 584000000], + endTime: [1786481102, 698000000], + value, + })), + } as MetricData; +} + +function resourceMetrics(metrics: MetricData[]): ResourceMetrics { + return { + resource: { attributes: {} }, + scopeMetrics: [{ scope: { name: "@opentelemetry/host-metrics" }, metrics }], + } as unknown as ResourceMetrics; +} + +async function exportThrough(exporter: BufferingMetricExporter, metrics: ResourceMetrics) { + exporter.export(metrics, () => {}); + await exporter.forceFlush(); +} + +function histogram(name: string, sums: number[]): MetricData { + return { + descriptor: { + name, + description: "", + unit: "ms", + type: InstrumentType.HISTOGRAM, + valueType: 1, + }, + aggregationTemporality: AggregationTemporality.DELTA, + dataPointType: DataPointType.HISTOGRAM, + dataPoints: sums.map((sum) => ({ + attributes: { "task.status": "completed" }, + startTime: [1786481102, 584000000], + endTime: [1786481102, 698000000], + value: { + min: 0, + max: 1, + sum, + count: 1, + buckets: { boundaries: [0, 1], counts: [0, 1, 0] }, + }, + })), + } as MetricData; +} + +describe("BufferingMetricExporter", () => { + it("forwards only the finite data points of a metric", async () => { + const inner = new RecordingMetricExporter(); + const exporter = new BufferingMetricExporter(inner, 30_000); + + await exportThrough(exporter, resourceMetrics([gauge("process.cpu.utilization", [NaN, 0.25])])); + + const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!; + expect(forwarded.dataPoints.map((dp) => dp.value)).toEqual([0.25]); + }); + + it("drops infinite data points as well as NaN", async () => { + const inner = new RecordingMetricExporter(); + const exporter = new BufferingMetricExporter(inner, 30_000); + + await exportThrough( + exporter, + resourceMetrics([gauge("process.cpu.utilization", [Infinity, -Infinity, 0.5])]) + ); + + const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!; + expect(forwarded.dataPoints.map((dp) => dp.value)).toEqual([0.5]); + }); + + it("forwards only the histogram data points with a finite sum", async () => { + const inner = new RecordingMetricExporter(); + const exporter = new BufferingMetricExporter(inner, 30_000); + + await exportThrough(exporter, resourceMetrics([histogram("task.duration", [NaN, 5])])); + + const forwarded = inner.exported[0]!.scopeMetrics[0]!.metrics[0]!; + expect(forwarded.dataPoints.map((dp) => (dp.value as { sum: number }).sum)).toEqual([5]); + }); + + describe("against a real otel collector", () => { + let collector: StartedOtelCollectorContainer; + + beforeAll(async () => { + collector = await new OtelCollectorContainer().start(); + }, 180_000); + + afterAll(async () => { + await collector?.stop(); + }); + + it("exports a batch containing a NaN gauge value without being rejected", async () => { + const exporter = new BufferingMetricExporter( + new OTLPMetricExporter({ url: `${collector.getOtlpHttpUrl()}/v1/metrics` }), + 30_000 + ); + + exporter.export(resourceMetrics([gauge("process.cpu.utilization", [NaN, 0.25])]), () => {}); + + await expect(exporter.forceFlush()).resolves.toBeUndefined(); + }); + }); +}); diff --git a/packages/core/src/v3/taskContext/otelProcessors.ts b/packages/core/src/v3/taskContext/otelProcessors.ts index 2a22a279d..2eb9c8aac 100644 --- a/packages/core/src/v3/taskContext/otelProcessors.ts +++ b/packages/core/src/v3/taskContext/otelProcessors.ts @@ -224,6 +224,37 @@ export class TaskContextMetricExporter implements PushMetricExporter { } } +function isFiniteDataPointValue(value: unknown): boolean { + if (typeof value === "number") { + return Number.isFinite(value); + } + + if (typeof value === "object" && value !== null) { + return (["sum", "min", "max"] as const).every((key) => { + const component = (value as Record)[key]; + return typeof component !== "number" || Number.isFinite(component); + }); + } + + return true; +} + +function dropNonFiniteDataPoints(metrics: ResourceMetrics): ResourceMetrics { + return { + ...metrics, + scopeMetrics: metrics.scopeMetrics.map((scope) => ({ + ...scope, + metrics: scope.metrics.map( + (metric) => + ({ + ...metric, + dataPoints: metric.dataPoints.filter((dp) => isFiniteDataPointValue(dp.value)), + }) as MetricData + ), + })), + }; +} + export class BufferingMetricExporter implements PushMetricExporter { selectAggregationTemporality?: (instrumentType: InstrumentType) => AggregationTemporality; selectAggregation?: (instrumentType: InstrumentType) => AggregationOption; @@ -245,7 +276,7 @@ export class BufferingMetricExporter implements PushMetricExporter { } export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void { - this._buffer.push(metrics); + this._buffer.push(dropNonFiniteDataPoints(metrics)); const now = Date.now(); if (now - this._lastFlushTime >= this._flushIntervalMs) { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 160c39b7b..36ebf893a 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1474,9 +1474,6 @@ importers: '@testcontainers/redis': specifier: ^11.14.0 version: 11.14.0 - '@trigger.dev/core': - specifier: workspace:* - version: link:../../packages/core std-env: specifier: ^3.9.0 version: 3.9.0 @@ -1940,6 +1937,9 @@ importers: '@epic-web/test-server': specifier: ^0.1.0 version: 0.1.0(bufferutil@4.0.9) + '@internal/testcontainers': + specifier: workspace:* + version: link:../../internal-packages/testcontainers '@trigger.dev/database': specifier: workspace:* version: link:../../internal-packages/database