From 93933a276ab52bf4f9d5c8622ea84a69c1617fc8 Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Fri, 14 Aug 2026 00:10:22 +0100 Subject: [PATCH 1/2] fix(core): stop custom metric exporters breaking the metrics export --- .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 | 205 ++++++++++++++++++ packages/core/src/v3/otel/tracingSDK.ts | 38 +++- .../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, 518 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 0000000000..2f43f51ff9 --- /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 465174fb95..ceb0b7b49d 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 291c1314c1..df0df141c6 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 e1cd3d25ae..da8c279808 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 0000000000..7df8db12ee --- /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 385b4113f5..a523eb5899 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 e2ee6f0642..75c95cd8fc 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 0000000000..0f5f1d8e9e --- /dev/null +++ b/packages/core/src/v3/otel/tracingSDK.test.ts @@ -0,0 +1,205 @@ +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 { + protected async onForceFlush(): Promise {} + + protected async onShutdown(): Promise { + throw new Error("reader shutdown failed"); + } +} + +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); + }); +}); diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 0b3a66a87b..4a1d5f487e 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,47 @@ 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() { + try { + await this._eachMetricReaderSerially("shut down", (reader) => reader.shutdown()); + } finally { + 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 0000000000..6d01eb72cf --- /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 0000000000..92d0271355 --- /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": String(value) }, + 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: { sum: String(sum) }, + 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 2a22a279db..2eb9c8aac1 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 160c39b7b2..36ebf893a2 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 From a4af9bc63fa655344981da81d8556e3b4855867a Mon Sep 17 00:00:00 2001 From: nicktrn <55853254+nicktrn@users.noreply.github.com> Date: Fri, 14 Aug 2026 00:23:22 +0100 Subject: [PATCH 2/2] fix(core): do not retry a metric reader that failed to shut down --- packages/core/src/v3/otel/tracingSDK.test.ts | 31 ++++++++++++++++++- packages/core/src/v3/otel/tracingSDK.ts | 7 ++--- .../src/v3/taskContext/otelProcessors.test.ts | 4 +-- 3 files changed, 34 insertions(+), 8 deletions(-) diff --git a/packages/core/src/v3/otel/tracingSDK.test.ts b/packages/core/src/v3/otel/tracingSDK.test.ts index 0f5f1d8e9e..3595fceff7 100644 --- a/packages/core/src/v3/otel/tracingSDK.test.ts +++ b/packages/core/src/v3/otel/tracingSDK.test.ts @@ -88,10 +88,13 @@ class FailingMetricReader extends MetricReader { } class FailingShutdownMetricReader extends MetricReader { + shutdownAttempts = 0; + protected async onForceFlush(): Promise {} protected async onShutdown(): Promise { - throw new Error("reader shutdown failed"); + this.shutdownAttempts++; + throw new Error(`reader shutdown failed (attempt ${this.shutdownAttempts})`); } } @@ -202,4 +205,30 @@ describe("TracingSDK shutdown", () => { 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 4a1d5f487e..0f4ea82a22 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -359,11 +359,8 @@ export class TracingSDK { } private async _shutdownMetricReadersSerially() { - try { - await this._eachMetricReaderSerially("shut down", (reader) => reader.shutdown()); - } finally { - await this._meterProvider.shutdown(); - } + await this._eachMetricReaderSerially("shut down", (reader) => reader.shutdown()); + await this._meterProvider.shutdown(); } private async _eachMetricReaderSerially( diff --git a/packages/core/src/v3/taskContext/otelProcessors.test.ts b/packages/core/src/v3/taskContext/otelProcessors.test.ts index 92d0271355..5bd1cd4e43 100644 --- a/packages/core/src/v3/taskContext/otelProcessors.test.ts +++ b/packages/core/src/v3/taskContext/otelProcessors.test.ts @@ -39,7 +39,7 @@ function gauge(name: string, values: number[]): MetricData { aggregationTemporality: AggregationTemporality.CUMULATIVE, dataPointType: DataPointType.GAUGE, dataPoints: values.map((value) => ({ - attributes: { "process.cpu.state": String(value) }, + attributes: { "process.cpu.state": "user" }, startTime: [1786481102, 584000000], endTime: [1786481102, 698000000], value, @@ -71,7 +71,7 @@ function histogram(name: string, sums: number[]): MetricData { aggregationTemporality: AggregationTemporality.DELTA, dataPointType: DataPointType.HISTOGRAM, dataPoints: sums.map((sum) => ({ - attributes: { sum: String(sum) }, + attributes: { "task.status": "completed" }, startTime: [1786481102, 584000000], endTime: [1786481102, 698000000], value: {