feat: metric aggregation temporality controls

CHANGELOG.md

@@ -6,6 +6,8 @@ All notable changes to this project will be documented in this file.
 
 ### :boom: Breaking Change
 
+* feat(metrics): metric readers and exporters now select aggregation temporality based on instrument type [#2902](https://github.com/open-telemetry/opentelemetry-js/pull/2902) @seemk
+
 ### :rocket: (Enhancement)
 
 * feat(ConsoleSpanExporter): export span links [#2917](https://github.com/open-telemetry/opentelemetry-js/pull/2917) @trentm

experimental/packages/opentelemetry-exporter-metrics-otlp-grpc/src/OTLPMetricExporter.ts

@@ -16,12 +16,11 @@
 
 import { otlpTypes } from '@opentelemetry/exporter-trace-otlp-http';
 import {
-  defaultExporterTemporality,
   defaultOptions,
   OTLPMetricExporterBase, OTLPMetricExporterOptions,
   toOTLPExportMetricServiceRequest
 } from '@opentelemetry/exporter-metrics-otlp-http';
-import { AggregationTemporality, ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
+import { ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
 import {
   OTLPGRPCExporterConfigNode,
   OTLPGRPCExporterNodeBase,
@@ -36,7 +35,6 @@ const DEFAULT_COLLECTOR_URL = 'localhost:4317';
 
 class OTLPMetricExporterProxy extends OTLPGRPCExporterNodeBase<ResourceMetrics,
   otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest> {
-  protected readonly _aggregationTemporality: AggregationTemporality;
 
   constructor(config: OTLPGRPCExporterConfigNode & OTLPMetricExporterOptions= defaultOptions) {
     super(config);
@@ -45,7 +43,6 @@ class OTLPMetricExporterProxy extends OTLPGRPCExporterNodeBase<ResourceMetrics,
     for (const [k, v] of Object.entries(headers)) {
       this.metadata.set(k, v);
     }
-    this._aggregationTemporality = config.aggregationTemporality ?? defaultExporterTemporality;
   }
 
   getServiceProtoPath(): string {
@@ -69,7 +66,6 @@ class OTLPMetricExporterProxy extends OTLPGRPCExporterNodeBase<ResourceMetrics,
   convert(metrics: ResourceMetrics[]): otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest {
     return toOTLPExportMetricServiceRequest(
       metrics[0],
-      this._aggregationTemporality,
       this
     );
   }

experimental/packages/opentelemetry-exporter-metrics-otlp-grpc/test/OTLPMetricExporter.test.ts

@@ -128,7 +128,7 @@ const testOTLPMetricExporter = (params: TestParams) =>
         url: 'grpcs://' + address,
         credentials,
         metadata: params.metadata,
-        aggregationTemporality: AggregationTemporality.CUMULATIVE
+        temporalityPreference: AggregationTemporality.CUMULATIVE
       });
 
       setUp();
@@ -182,15 +182,15 @@ const testOTLPMetricExporter = (params: TestParams) =>
           headers: {
             foo: 'bar',
           },
-          aggregationTemporality: AggregationTemporality.CUMULATIVE
+          temporalityPreference: AggregationTemporality.CUMULATIVE
         });
         const args = warnStub.args[0];
         assert.strictEqual(args[0], 'Headers cannot be set when using grpc');
       });
       it('should warn about path in url', () => {
         collectorExporter = new OTLPMetricExporter({
           url: `http://${address}/v1/metrics`,
-          aggregationTemporality: AggregationTemporality.CUMULATIVE
+          temporalityPreference: AggregationTemporality.CUMULATIVE
         });
         const args = warnStub.args[0];
         assert.strictEqual(
@@ -264,7 +264,7 @@ describe('OTLPMetricExporter - node (getDefaultUrl)', () => {
     const url = 'http://foo.bar.com';
     const collectorExporter = new OTLPMetricExporter({
       url,
-      aggregationTemporality: AggregationTemporality.CUMULATIVE
+      temporalityPreference: AggregationTemporality.CUMULATIVE
     });
     setTimeout(() => {
       assert.strictEqual(collectorExporter._otlpExporter.url, 'foo.bar.com');
@@ -309,7 +309,7 @@ describe('when configuring via environment', () => {
     envSource.OTEL_EXPORTER_OTLP_METRICS_HEADERS = 'foo=boo';
     const collectorExporter = new OTLPMetricExporter({
       metadata,
-      aggregationTemporality: AggregationTemporality.CUMULATIVE
+      temporalityPreference: AggregationTemporality.CUMULATIVE
     });
     assert.deepStrictEqual(collectorExporter._otlpExporter.metadata?.get('foo'), ['boo']);
     assert.deepStrictEqual(collectorExporter._otlpExporter.metadata?.get('bar'), ['foo']);

experimental/packages/opentelemetry-exporter-metrics-otlp-grpc/test/metricsHelper.ts

@@ -20,9 +20,18 @@ import { Resource } from '@opentelemetry/resources';
 import * as assert from 'assert';
 import * as grpc from '@grpc/grpc-js';
 import { VERSION } from '@opentelemetry/core';
-import { ExplicitBucketHistogramAggregation, MeterProvider, MetricReader } from '@opentelemetry/sdk-metrics-base';
+import {
+  AggregationTemporality,
+  ExplicitBucketHistogramAggregation,
+  MeterProvider,
+  MetricReader,
+} from '@opentelemetry/sdk-metrics-base';
+
+class TestMetricReader extends MetricReader {
+  selectAggregationTemporality() {
+    return AggregationTemporality.CUMULATIVE;
+  }
 
-export class TestMetricReader extends MetricReader {
   protected onForceFlush(): Promise<void> {
     return Promise.resolve(undefined);
   }

experimental/packages/opentelemetry-exporter-metrics-otlp-http/src/OTLPMetricExporterBase.ts

@@ -15,22 +15,51 @@
  */
 
 import { ExportResult } from '@opentelemetry/core';
-import { AggregationTemporality, PushMetricExporter, ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
+import {
+  AggregationTemporality,
+  AggregationTemporalitySelector,
+  InstrumentType,
+  PushMetricExporter,
+  ResourceMetrics
+} from '@opentelemetry/sdk-metrics-base';
 import { otlpTypes } from '@opentelemetry/exporter-trace-otlp-http';
 import { defaultOptions, OTLPMetricExporterOptions } from './OTLPMetricExporterOptions';
 import { OTLPExporterBase } from '@opentelemetry/otlp-exporter-base';
 
+export const CumulativeTemporalitySelector: AggregationTemporalitySelector = () => AggregationTemporality.CUMULATIVE;
+
+export const DeltaTemporalitySelector: AggregationTemporalitySelector = (instrumentType: InstrumentType) => {
+  switch (instrumentType) {
+    case InstrumentType.COUNTER:
+    case InstrumentType.OBSERVABLE_COUNTER:
+    case InstrumentType.HISTOGRAM:
+    case InstrumentType.OBSERVABLE_GAUGE:
+      return AggregationTemporality.DELTA;
+    case InstrumentType.UP_DOWN_COUNTER:
+    case InstrumentType.OBSERVABLE_UP_DOWN_COUNTER:
+      return AggregationTemporality.CUMULATIVE;
+  }
+};
+
+function chooseTemporalitySelector(temporalityPreference?: AggregationTemporality): AggregationTemporalitySelector {
+  if (temporalityPreference === AggregationTemporality.DELTA) {
+    return DeltaTemporalitySelector;
+  }
+
+  return CumulativeTemporalitySelector;
+}
+
 export class OTLPMetricExporterBase<T extends OTLPExporterBase<OTLPMetricExporterOptions,
   ResourceMetrics,
   otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest>>
 implements PushMetricExporter {
   public _otlpExporter: T;
-  protected _preferredAggregationTemporality: AggregationTemporality;
+  protected _aggregationTemporalitySelector: AggregationTemporalitySelector;
 
   constructor(exporter: T,
     config: OTLPMetricExporterOptions = defaultOptions) {
     this._otlpExporter = exporter;
-    this._preferredAggregationTemporality = config.aggregationTemporality ?? AggregationTemporality.CUMULATIVE;
+    this._aggregationTemporalitySelector = chooseTemporalitySelector(config.temporalityPreference);
   }
 
   export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void {
@@ -45,8 +74,8 @@ implements PushMetricExporter {
     return Promise.resolve();
   }
 
-  getPreferredAggregationTemporality(): AggregationTemporality {
-    return this._preferredAggregationTemporality;
+  selectAggregationTemporality(instrumentType: InstrumentType): AggregationTemporality {
+    return this._aggregationTemporalitySelector(instrumentType);
   }
 
 }

experimental/packages/opentelemetry-exporter-metrics-otlp-http/src/OTLPMetricExporterOptions.ts

@@ -18,7 +18,7 @@ import { AggregationTemporality } from '@opentelemetry/sdk-metrics-base';
 import { OTLPExporterConfigBase } from '@opentelemetry/otlp-exporter-base';
 
 export interface OTLPMetricExporterOptions extends OTLPExporterConfigBase {
-  aggregationTemporality?: AggregationTemporality
+  temporalityPreference?: AggregationTemporality
 }
 export const defaultExporterTemporality = AggregationTemporality.CUMULATIVE;
-export const defaultOptions = {aggregationTemporality: defaultExporterTemporality};
+export const defaultOptions = {temporalityPreference: defaultExporterTemporality};

experimental/packages/opentelemetry-exporter-metrics-otlp-http/src/platform/browser/OTLPMetricExporter.ts

@@ -14,13 +14,13 @@
  * limitations under the License.
  */
 
-import { AggregationTemporality, ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
+import { ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
 import {
   otlpTypes
 } from '@opentelemetry/exporter-trace-otlp-http';
 import { toOTLPExportMetricServiceRequest } from '../../transformMetrics';
 import { baggageUtils, getEnv } from '@opentelemetry/core';
-import { defaultExporterTemporality, defaultOptions, OTLPMetricExporterOptions } from '../../OTLPMetricExporterOptions';
+import { defaultOptions, OTLPMetricExporterOptions } from '../../OTLPMetricExporterOptions';
 import { OTLPMetricExporterBase } from '../../OTLPMetricExporterBase';
 import {
   appendResourcePathToUrlIfNotPresent,
@@ -33,7 +33,6 @@ const DEFAULT_COLLECTOR_URL = `http://localhost:4318${DEFAULT_COLLECTOR_RESOURCE
 
 class OTLPExporterBrowserProxy extends OTLPExporterBrowserBase<ResourceMetrics,
   otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest> {
-  protected readonly _aggregationTemporality: AggregationTemporality;
 
   constructor(config: OTLPMetricExporterOptions & OTLPExporterConfigBase = defaultOptions) {
     super(config);
@@ -43,7 +42,6 @@ class OTLPExporterBrowserProxy extends OTLPExporterBrowserBase<ResourceMetrics,
         getEnv().OTEL_EXPORTER_OTLP_METRICS_HEADERS
       )
     );
-    this._aggregationTemporality = config.aggregationTemporality ?? defaultExporterTemporality;
   }
 
   getDefaultUrl(config: OTLPExporterConfigBase): string {
@@ -61,7 +59,6 @@ class OTLPExporterBrowserProxy extends OTLPExporterBrowserBase<ResourceMetrics,
   ): otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest {
     return toOTLPExportMetricServiceRequest(
       metrics[0],
-      this._aggregationTemporality,
       this
     );
   }

experimental/packages/opentelemetry-exporter-metrics-otlp-http/src/platform/node/OTLPMetricExporter.ts

@@ -14,13 +14,13 @@
  * limitations under the License.
  */
 
-import { ResourceMetrics, AggregationTemporality } from '@opentelemetry/sdk-metrics-base';
+import { ResourceMetrics } from '@opentelemetry/sdk-metrics-base';
 import {
   otlpTypes,
 } from '@opentelemetry/exporter-trace-otlp-http';
 import { toOTLPExportMetricServiceRequest } from '../../transformMetrics';
 import { getEnv, baggageUtils} from '@opentelemetry/core';
-import { defaultExporterTemporality, defaultOptions, OTLPMetricExporterOptions } from '../../OTLPMetricExporterOptions';
+import { defaultOptions, OTLPMetricExporterOptions } from '../../OTLPMetricExporterOptions';
 import { OTLPMetricExporterBase } from '../../OTLPMetricExporterBase';
 import {
   appendResourcePathToUrlIfNotPresent,
@@ -33,7 +33,6 @@ const DEFAULT_COLLECTOR_URL = `http://localhost:4318${DEFAULT_COLLECTOR_RESOURCE
 
 class OTLPExporterNodeProxy extends OTLPExporterNodeBase<ResourceMetrics,
   otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest> {
-  protected readonly _aggregationTemporality: AggregationTemporality;
 
   constructor(config: OTLPExporterNodeConfigBase & OTLPMetricExporterOptions = defaultOptions) {
     super(config);
@@ -43,15 +42,13 @@ class OTLPExporterNodeProxy extends OTLPExporterNodeBase<ResourceMetrics,
         getEnv().OTEL_EXPORTER_OTLP_METRICS_HEADERS
       )
     );
-    this._aggregationTemporality = config.aggregationTemporality ?? defaultExporterTemporality;
   }
 
   convert(
     metrics: ResourceMetrics[]
   ): otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest {
     return toOTLPExportMetricServiceRequest(
       metrics[0],
-      this._aggregationTemporality,
       this
     );
   }

experimental/packages/opentelemetry-exporter-metrics-otlp-http/src/transformMetrics.ts

@@ -113,21 +113,19 @@ export function toHistogramDataPoints(
  */
 export function toCollectorMetric(
   metricData: MetricData,
-  aggregationTemporality: AggregationTemporality
 ): otlpTypes.opentelemetryProto.metrics.v1.Metric {
   const metricCollector: otlpTypes.opentelemetryProto.metrics.v1.Metric = {
     name: metricData.descriptor.name,
     description: metricData.descriptor.description,
     unit: metricData.descriptor.unit,
   };
-
   if (metricData.dataPointType === DataPointType.SINGULAR) {
     const result = {
       dataPoints: toSingularDataPoints(metricData),
       isMonotonic:
         metricData.descriptor.type === InstrumentType.COUNTER ||
         metricData.descriptor.type === InstrumentType.OBSERVABLE_COUNTER,
-      aggregationTemporality: toAggregationTemporality(aggregationTemporality),
+      aggregationTemporality: toAggregationTemporality(metricData.aggregationTemporality),
     };
 
     if (
@@ -152,7 +150,7 @@ export function toCollectorMetric(
   } else if (metricData.dataPointType === DataPointType.HISTOGRAM) {
     const result = {
       dataPoints: toHistogramDataPoints(metricData),
-      aggregationTemporality: toAggregationTemporality(aggregationTemporality)
+      aggregationTemporality: toAggregationTemporality(metricData.aggregationTemporality)
     };
     if (metricData.descriptor.valueType === ValueType.INT) {
       metricCollector.intHistogram = result;
@@ -174,7 +172,6 @@ export function toCollectorMetric(
  */
 export function toOTLPExportMetricServiceRequest<T extends OTLPExporterConfigBase>(
   metrics: ResourceMetrics,
-  aggregationTemporality: AggregationTemporality,
   collectorExporterBase: OTLPExporterBase<T,
     ResourceMetrics,
     otlpTypes.opentelemetryProto.collector.metrics.v1.ExportMetricsServiceRequest>
@@ -187,7 +184,6 @@ export function toOTLPExportMetricServiceRequest<T extends OTLPExporterConfigBas
     resourceMetrics: toCollectorResourceMetrics(
       metrics,
       additionalAttributes,
-      aggregationTemporality
     ),
   };
 }
@@ -196,15 +192,14 @@ export function toOTLPExportMetricServiceRequest<T extends OTLPExporterConfigBas
  * Convert to InstrumentationLibraryMetrics
  * @param instrumentationLibrary
  * @param metrics
- * @param aggregationTemporality
+ * @param temporalitySelector
  */
 function toCollectorInstrumentationLibraryMetrics(
   instrumentationLibrary: core.InstrumentationLibrary,
   metrics: MetricData[],
-  aggregationTemporality: AggregationTemporality
 ): otlpTypes.opentelemetryProto.metrics.v1.InstrumentationLibraryMetrics {
   return {
-    metrics: metrics.map(metric => toCollectorMetric(metric, aggregationTemporality)),
+    metrics: metrics.map(metric => toCollectorMetric(metric)),
     instrumentationLibrary,
   };
 }
@@ -218,15 +213,13 @@ function toCollectorInstrumentationLibraryMetrics(
 function toCollectorResourceMetrics(
   resourceMetrics: ResourceMetrics,
   baseAttributes: SpanAttributes,
-  aggregationTemporality: AggregationTemporality
 ): otlpTypes.opentelemetryProto.metrics.v1.ResourceMetrics[] {
   return [{
     resource: toCollectorResource(resourceMetrics.resource, baseAttributes),
     instrumentationLibraryMetrics: Array.from(resourceMetrics.instrumentationLibraryMetrics.map(
       instrumentationLibraryMetrics => toCollectorInstrumentationLibraryMetrics(
         instrumentationLibraryMetrics.instrumentationLibrary,
         instrumentationLibraryMetrics.metrics,
-        aggregationTemporality
       )))
   }];
 }

experimental/packages/opentelemetry-exporter-metrics-otlp-http/test/browser/CollectorMetricExporter.test.ts

@@ -94,7 +94,7 @@ describe('OTLPMetricExporter - web', () => {
       beforeEach(() => {
         collectorExporter = new OTLPMetricExporter({
           url: 'http://foo.bar.com',
-          aggregationTemporality: AggregationTemporality.CUMULATIVE
+          temporalityPreference: AggregationTemporality.CUMULATIVE
         });
       });
       it('should successfully send metrics using sendBeacon', done => {
@@ -202,7 +202,7 @@ describe('OTLPMetricExporter - web', () => {
         (window.navigator as any).sendBeacon = false;
         collectorExporter = new OTLPMetricExporter({
           url: 'http://foo.bar.com',
-          aggregationTemporality: AggregationTemporality.CUMULATIVE
+          temporalityPreference: AggregationTemporality.CUMULATIVE
         });
         // Overwrites the start time to make tests consistent
         Object.defineProperty(collectorExporter, '_startTime', {
@@ -338,7 +338,7 @@ describe('OTLPMetricExporter - web', () => {
     beforeEach(() => {
       collectorExporterConfig = {
         headers: customHeaders,
-        aggregationTemporality: AggregationTemporality.CUMULATIVE
+        temporalityPreference: AggregationTemporality.CUMULATIVE
       };
       server = sinon.fakeServer.create();
     });
@@ -430,7 +430,7 @@ describe('when configuring via environment', () => {
     envSource.OTEL_EXPORTER_OTLP_HEADERS = 'foo=bar';
     const collectorExporter = new OTLPMetricExporter({
       headers: {},
-      aggregationTemporality: AggregationTemporality.CUMULATIVE
+      temporalityPreference: AggregationTemporality.CUMULATIVE
     });
     assert.strictEqual(collectorExporter['_otlpExporter']['_headers'].foo, 'bar');
     envSource.OTEL_EXPORTER_OTLP_HEADERS = '';
@@ -440,7 +440,7 @@ describe('when configuring via environment', () => {
     envSource.OTEL_EXPORTER_OTLP_METRICS_HEADERS = 'foo=boo';
     const collectorExporter = new OTLPMetricExporter({
       headers: {},
-      aggregationTemporality: AggregationTemporality.CUMULATIVE
+      temporalityPreference: AggregationTemporality.CUMULATIVE
     });
     assert.strictEqual(collectorExporter['_otlpExporter']['_headers'].foo, 'boo');
     assert.strictEqual(collectorExporter['_otlpExporter']['_headers'].bar, 'foo');