diff --git a/.changeset/otel-metrics-reader-temporality.md b/.changeset/otel-metrics-reader-temporality.md new file mode 100644 index 00000000000..843c165add0 --- /dev/null +++ b/.changeset/otel-metrics-reader-temporality.md @@ -0,0 +1,14 @@ +--- +"@effect/opentelemetry": patch +--- + +`OtelMetrics` now honors the aggregation temporality preference of the registered `MetricReader`. + +When no explicit `temporality` option is passed to `OtelMetrics.layer` or +`OtelMetrics.makeProducer`, the metric producer queries the reader's +`selectAggregationTemporality(instrumentType)` for each produced metric, so +exporters configured with a temporality preference (e.g. +`OTLPMetricExporter({ temporalityPreference: DELTA })`) are no longer silently +overridden with cumulative data points. An explicit `temporality` option still +takes precedence, and cumulative remains the default when the reader expresses +no preference. diff --git a/packages/opentelemetry/src/OtelMetrics.ts b/packages/opentelemetry/src/OtelMetrics.ts index da76f6b7c61..1d279e0b459 100644 --- a/packages/opentelemetry/src/OtelMetrics.ts +++ b/packages/opentelemetry/src/OtelMetrics.ts @@ -26,9 +26,12 @@ import { Resource } from "./Resource.ts" * **Details** * * `cumulative` reports total since a fixed start time. Each data point depends - * on all previous measurements. This is the default behavior. `delta` reports - * changes since the last export. Each interval is independent with no - * dependency on previous measurements. + * on all previous measurements. `delta` reports changes since the last export. + * Each interval is independent with no dependency on previous measurements. + * + * When no preference is configured, the producer follows the temporality + * preference of the `MetricReader` it is registered with, falling back to + * cumulative. * * @category models * @since 4.0.0 @@ -45,9 +48,11 @@ export type TemporalityPreference = "cumulative" | "delta" * * **Details** * - * Requires the current OpenTelemetry `Resource`, captures the current Effect - * context, and uses cumulative temporality by default. Pass `"delta"` for - * interval-based values. + * Requires the current OpenTelemetry `Resource` and captures the current + * Effect context. When no temporality is passed, the producer follows the + * temporality preference of the `MetricReader` it is registered with via + * {@link registerProducer} or {@link layer}, falling back to cumulative. Pass + * `"cumulative"` or `"delta"` to override the reader's preference. * * @see {@link registerProducer} for attaching a producer to metric readers * @see {@link layer} for creating and registering a producer in a scoped layer @@ -79,7 +84,15 @@ export const registerProducer = ( Effect.sync(() => { const reader = metricReader() const readers: Array = Array.isArray(reader) ? reader : [reader] as any - readers.forEach((reader) => reader.setMetricProducer(self instanceof MetricProducerImpl ? self.fork() : self)) + readers.forEach((reader) => { + if (self instanceof MetricProducerImpl) { + const producer = self.fork() + producer.reader = reader + reader.setMetricProducer(producer) + } else { + reader.setMetricProducer(self) + } + }) return readers }), (readers) => diff --git a/packages/opentelemetry/src/internal/metrics.ts b/packages/opentelemetry/src/internal/metrics.ts index 3fd560a7a7f..f79c2bcd9c9 100644 --- a/packages/opentelemetry/src/internal/metrics.ts +++ b/packages/opentelemetry/src/internal/metrics.ts @@ -6,7 +6,8 @@ import type { Histogram, MetricCollectOptions, MetricData, - MetricProducer + MetricProducer, + MetricReader } from "@opentelemetry/sdk-metrics" import { AggregationTemporality, DataPointType, InstrumentType } from "@opentelemetry/sdk-metrics" import type { InstrumentDescriptor } from "@opentelemetry/sdk-metrics/build/src/InstrumentDescriptor.js" @@ -39,7 +40,10 @@ interface PreviousSummaryState { export class MetricProducerImpl implements MetricProducer { resource: Resources.Resource context: Context.Context - temporality: Metrics.TemporalityPreference + temporality: Metrics.TemporalityPreference | undefined + // set by registerProducer, so collect() can follow the reader's temporality + // preference when no explicit temporality was configured + reader: MetricReader | undefined startTimes: Map startTimeNanos: HrTime previousExportTimeNanos: HrTime @@ -51,11 +55,12 @@ export class MetricProducerImpl implements MetricProducer { constructor( resource: Resources.Resource, context: Context.Context, - temporality: Metrics.TemporalityPreference = "cumulative" + temporality?: Metrics.TemporalityPreference ) { this.resource = resource this.context = context this.temporality = temporality + this.reader = undefined this.startTimes = new Map() this.startTimeNanos = currentHrTime() this.previousExportTimeNanos = this.startTimeNanos @@ -77,6 +82,20 @@ export class MetricProducerImpl implements MetricProducer { return hrTime } + // an explicitly configured temporality wins, then the registered reader's + // per-instrument-type preference, then the OpenTelemetry default of + // cumulative + selectTemporality(instrumentType: InstrumentType): AggregationTemporality { + if (this.temporality !== undefined) { + return this.temporality === "delta" + ? AggregationTemporality.DELTA + : AggregationTemporality.CUMULATIVE + } else if (this.reader !== undefined) { + return this.reader.selectAggregationTemporality(instrumentType) + } + return AggregationTemporality.CUMULATIVE + } + collect(_options?: MetricCollectOptions): Promise { const snapshot = Metric.snapshotUnsafe(this.context) const hrTimeNow = currentHrTime() @@ -87,16 +106,13 @@ export class MetricProducerImpl implements MetricProducer { metricDataByName.set(data.descriptor.name, data) } - const isDelta = this.temporality === "delta" - const aggregationTemporality = isDelta - ? AggregationTemporality.DELTA - : AggregationTemporality.CUMULATIVE - const intervalStartTime = isDelta - ? this.previousExportTimeNanos - : this.startTimeNanos - for (let i = 0, len = snapshot.length; i < len; i++) { const state = snapshot[i] + const aggregationTemporality = this.selectTemporality(instrumentTypeFromSnapshot(state)) + const isDelta = aggregationTemporality === AggregationTemporality.DELTA + const intervalStartTime = isDelta + ? this.previousExportTimeNanos + : this.startTimeNanos const attributes = state.attributes ? Arr.reduce(Object.entries(state.attributes), {} as Record, (acc, [key, value]) => { Rec.assignProperty(acc, key, String(value)) @@ -397,9 +413,7 @@ export class MetricProducerImpl implements MetricProducer { } // Update the previous export time for delta calculations - if (isDelta) { - this.previousExportTimeNanos = hrTimeNow - } + this.previousExportTimeNanos = hrTimeNow return Promise.resolve({ resourceMetrics: { diff --git a/packages/opentelemetry/test/OtelMetrics.test.ts b/packages/opentelemetry/test/OtelMetrics.test.ts index df8947529a9..f2c087cfdec 100644 --- a/packages/opentelemetry/test/OtelMetrics.test.ts +++ b/packages/opentelemetry/test/OtelMetrics.test.ts @@ -3,13 +3,22 @@ import * as OtelMetrics from "@effect/opentelemetry/OtelMetrics" import { assert, describe, it } from "@effect/vitest" import { ValueType } from "@opentelemetry/api" import { resourceFromAttributes } from "@opentelemetry/resources" -import { MetricReader } from "@opentelemetry/sdk-metrics" +import { AggregationTemporality, InstrumentType, MetricReader } from "@opentelemetry/sdk-metrics" import * as Effect from "effect/Effect" import * as Metric from "effect/Metric" const findMetric = (metrics: any, name: string) => metrics.resourceMetrics.scopeMetrics[0].metrics.find((_: any) => _.descriptor.name === name) +class TestReader extends MetricReader { + protected onShutdown(): Promise { + return Promise.resolve() + } + protected onForceFlush(): Promise { + return Promise.resolve() + } +} + describe("Metrics", () => { it.effect("gauge", () => Effect.gen(function*() { @@ -363,4 +372,74 @@ describe("Metrics", () => { const secondValue = (secondResult.resourceMetrics.scopeMetrics[0]!.metrics[0] as any).dataPoints[0].value assert.deepStrictEqual([firstValue, secondValue], [1, 1]) }).pipe(Effect.provideService(Metric.MetricRegistry, new Map()))) + + it.effect("follows the reader's temporality preference when none is configured", () => + Effect.gen(function*() { + const services = yield* Effect.context() + const producer = new internal.MetricProducerImpl(resourceFromAttributes({}), services) + const reader = new TestReader({ + aggregationTemporalitySelector: () => AggregationTemporality.DELTA + }) + yield* OtelMetrics.registerProducer(producer, () => reader) + const counter = Metric.counter("delta.requests", { incremental: true }) + + yield* Metric.update(counter, 3) + const first = findMetric(yield* Effect.promise(() => reader.collect()), "delta.requests") + yield* Metric.update(counter, 2) + const second = findMetric(yield* Effect.promise(() => reader.collect()), "delta.requests") + + assert.strictEqual(first.aggregationTemporality, AggregationTemporality.DELTA) + assert.strictEqual(first.dataPoints[0].value, 3) + assert.strictEqual(second.aggregationTemporality, AggregationTemporality.DELTA) + assert.strictEqual(second.dataPoints[0].value, 2) + }).pipe(Effect.provideService(Metric.MetricRegistry, new Map()))) + + it.effect("selects the reader's temporality per instrument type", () => + Effect.gen(function*() { + const services = yield* Effect.context() + const producer = new internal.MetricProducerImpl(resourceFromAttributes({}), services) + const reader = new TestReader({ + aggregationTemporalitySelector: (instrumentType) => + instrumentType === InstrumentType.COUNTER + ? AggregationTemporality.DELTA + : AggregationTemporality.CUMULATIVE + }) + yield* OtelMetrics.registerProducer(producer, () => reader) + const requests = Metric.counter("typed.requests", { incremental: true }) + const inflight = Metric.counter("typed.inflight") + + yield* Metric.update(requests, 3) + yield* Metric.update(inflight, 3) + yield* Effect.promise(() => reader.collect()) + yield* Metric.update(requests, 2) + yield* Metric.update(inflight, 2) + const result = yield* Effect.promise(() => reader.collect()) + + const counter = findMetric(result, "typed.requests") + const upDownCounter = findMetric(result, "typed.inflight") + assert.strictEqual(counter.aggregationTemporality, AggregationTemporality.DELTA) + assert.strictEqual(counter.dataPoints[0].value, 2) + assert.strictEqual(upDownCounter.aggregationTemporality, AggregationTemporality.CUMULATIVE) + assert.strictEqual(upDownCounter.dataPoints[0].value, 5) + }).pipe(Effect.provideService(Metric.MetricRegistry, new Map()))) + + it.effect("an explicit temporality overrides the reader's preference", () => + Effect.gen(function*() { + const services = yield* Effect.context() + const producer = new internal.MetricProducerImpl(resourceFromAttributes({}), services, "cumulative") + const reader = new TestReader({ + aggregationTemporalitySelector: () => AggregationTemporality.DELTA + }) + yield* OtelMetrics.registerProducer(producer, () => reader) + const counter = Metric.counter("override.requests", { incremental: true }) + + yield* Metric.update(counter, 3) + yield* Effect.promise(() => reader.collect()) + yield* Metric.update(counter, 2) + const result = yield* Effect.promise(() => reader.collect()) + + const metric = findMetric(result, "override.requests") + assert.strictEqual(metric.aggregationTemporality, AggregationTemporality.CUMULATIVE) + assert.strictEqual(metric.dataPoints[0].value, 5) + }).pipe(Effect.provideService(Metric.MetricRegistry, new Map()))) })