Honor MetricReader temporality preference in OtelMetrics

.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.

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<MetricReader> = 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) =>

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<never>
-  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<string, HrTime>
   startTimeNanos: HrTime
   previousExportTimeNanos: HrTime
@@ -51,11 +55,12 @@ export class MetricProducerImpl implements MetricProducer {
   constructor(
     resource: Resources.Resource,
     context: Context.Context<never>,
-    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<CollectionResult> {
     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<string, string>, (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: {

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<void> {
+    return Promise.resolve()
+  }
+  protected onForceFlush(): Promise<void> {
+    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<never>()
+      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<never>()
+      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<never>()
+      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())))
 })