Aggregation Temporality Support

packages/opentelemetry-exporter-collector/test/common/transformMetrics.test.ts

@@ -123,7 +123,7 @@ describe('transformMetrics', () => {
       ensureSumObserverIsCorrect(
         transform.toCollectorMetric(sumObserverMetric, 1592602232694000000),
         hrTimeToNanoseconds(sumObserverMetric.aggregator.toPoint().timestamp),
-        3
+        -1
       );
 
       // collect 3 times

packages/opentelemetry-exporter-prometheus/test/ExactProcessor.ts

@@ -15,14 +15,13 @@
  */
 import {
   Aggregator,
+  BasicProcessor,
   MetricDescriptor,
-  MetricRecord,
-  Processor,
 } from '@opentelemetry/metrics';
 
 type Constructor<T, R extends Aggregator> = new (...args: T[]) => R;
 
-export class ExactProcessor<T, R extends Aggregator> extends Processor {
+export class ExactProcessor<T, R extends Aggregator> extends BasicProcessor {
   private readonly args: ConstructorParameters<Constructor<T, R>>;
   public aggregators: R[] = [];
 
@@ -39,11 +38,4 @@ export class ExactProcessor<T, R extends Aggregator> extends Processor {
     this.aggregators.push(aggregator);
     return aggregator;
   }
-
-  process(record: MetricRecord): void {
-    const labels = Object.keys(record.labels)
-      .map(k => `${k}=${record.labels[k]}`)
-      .join(',');
-    this._batchMap.set(record.descriptor.name + labels, record);
-  }
 }

packages/opentelemetry-exporter-prometheus/test/PrometheusExporter.test.ts

@@ -203,35 +203,28 @@ describe('PrometheusExporter', () => {
       boundCounter.add(10);
       meter.collect().then(() => {
         exporter.export(meter.getProcessor().checkPointSet(), () => {
-          // TODO: Remove this special case once the PR is ready.
-          // This is to test the special case where counters are destroyed
-          // and recreated in the exporter in order to get around prom-client's
-          // aggregation and use ours.
-          boundCounter.add(10);
-          exporter.export(meter.getProcessor().checkPointSet(), () => {
-            http
-              .get('http://localhost:9464/metrics', res => {
-                res.on('data', chunk => {
-                  const body = chunk.toString();
-                  const lines = body.split('\n');
+          http
+            .get('http://localhost:9464/metrics', res => {
+              res.on('data', chunk => {
+                const body = chunk.toString();
+                const lines = body.split('\n');
 
-                  assert.strictEqual(
-                    lines[0],
-                    '# HELP counter a test description'
-                  );
+                assert.strictEqual(
+                  lines[0],
+                  '# HELP counter a test description'
+                );
 
-                  assert.deepStrictEqual(lines, [
-                    '# HELP counter a test description',
-                    '# TYPE counter counter',
-                    `counter{key1="labelValue1"} 20 ${mockedHrTimeMs}`,
-                    '',
-                  ]);
+                assert.deepStrictEqual(lines, [
+                  '# HELP counter a test description',
+                  '# TYPE counter counter',
+                  `counter{key1="labelValue1"} 10 ${mockedHrTimeMs}`,
+                  '',
+                ]);
 
-                  done();
-                });
-              })
-              .on('error', errorHandler(done));
-          });
+                done();
+              });
+            })
+            .on('error', errorHandler(done));
         });
       });
     });

packages/opentelemetry-metrics/src/Accumulator.ts

@@ -0,0 +1,20 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { Labels } from '@opentelemetry/api';
+
+export interface Accumulator {
+  update(accumulationKey: string, labels: Labels, value: number): void;
+}

packages/opentelemetry-metrics/src/BaseObserverMetric.ts

@@ -16,6 +16,7 @@
 import * as api from '@opentelemetry/api';
 import { InstrumentationLibrary } from '@opentelemetry/core';
 import { Resource } from '@opentelemetry/resources';
+import { Accumulator } from './Accumulator';
 import { BoundObserver } from './BoundInstrument';
 import { Processor } from './export/Processor';
 import { MetricKind, MetricRecord } from './export/types';
@@ -28,31 +29,42 @@ const NOOP_CALLBACK = () => {};
  * This is a SDK implementation of Base Observer Metric.
  * All observers should extend this class
  */
-export abstract class BaseObserverMetric
-  extends Metric<BoundObserver>
+export abstract class BaseObserverMetric extends Metric<BoundObserver>
   implements api.BaseObserver {
   protected _callback: (observerResult: api.ObserverResult) => unknown;
 
   constructor(
     name: string,
     options: api.MetricOptions,
-    private readonly _processor: Processor,
+    processor: Processor,
     resource: Resource,
     metricKind: MetricKind,
     instrumentationLibrary: InstrumentationLibrary,
     callback?: (observerResult: api.ObserverResult) => unknown
   ) {
-    super(name, options, metricKind, resource, instrumentationLibrary);
+    super(
+      name,
+      options,
+      metricKind,
+      processor,
+      resource,
+      instrumentationLibrary
+    );
     this._callback = callback || NOOP_CALLBACK;
   }
 
-  protected _makeInstrument(labels: api.Labels): BoundObserver {
+  protected _makeInstrument(
+    accumulationKey: string,
+    labels: api.Labels,
+    accumulator: Accumulator
+  ): BoundObserver {
     return new BoundObserver(
+      accumulationKey,
       labels,
-      this._disabled,
-      this._valueType,
+      accumulator,
       this._logger,
-      this._processor.aggregatorFor(this._descriptor)
+      this._disabled,
+      this._valueType
     );
   }
 

packages/opentelemetry-metrics/src/BatchObserverMetric.ts

@@ -17,6 +17,7 @@
 import * as api from '@opentelemetry/api';
 import { InstrumentationLibrary } from '@opentelemetry/core';
 import { Resource } from '@opentelemetry/resources';
+import { Accumulator } from './Accumulator';
 import { BatchObserverResult } from './BatchObserverResult';
 import { BoundObserver } from './BoundInstrument';
 import { Processor } from './export/Processor';
@@ -27,16 +28,15 @@ const NOOP_CALLBACK = () => {};
 const MAX_TIMEOUT_UPDATE_MS = 500;
 
 /** This is a SDK implementation of Batch Observer Metric. */
-export class BatchObserverMetric
-  extends Metric<BoundObserver>
+export class BatchObserverMetric extends Metric<BoundObserver>
   implements api.BatchObserver {
   private _callback: (observerResult: api.BatchObserverResult) => void;
   private _maxTimeoutUpdateMS: number;
 
   constructor(
     name: string,
     options: api.BatchMetricOptions,
-    private readonly _processor: Processor,
+    processor: Processor,
     resource: Resource,
     instrumentationLibrary: InstrumentationLibrary,
     callback?: (observerResult: api.BatchObserverResult) => void
@@ -45,6 +45,7 @@ export class BatchObserverMetric
       name,
       options,
       MetricKind.BATCH_OBSERVER,
+      processor,
       resource,
       instrumentationLibrary
     );
@@ -53,13 +54,18 @@ export class BatchObserverMetric
     this._callback = callback || NOOP_CALLBACK;
   }
 
-  protected _makeInstrument(labels: api.Labels): BoundObserver {
+  protected _makeInstrument(
+    accumulationKey: string,
+    labels: api.Labels,
+    accumulator: Accumulator
+  ): BoundObserver {
     return new BoundObserver(
+      accumulationKey,
       labels,
-      this._disabled,
-      this._valueType,
+      accumulator,
       this._logger,
-      this._processor.aggregatorFor(this._descriptor)
+      this._disabled,
+      this._valueType
     );
   }
 

packages/opentelemetry-metrics/src/BoundInstrument.ts

@@ -15,26 +15,21 @@
  */
 
 import * as api from '@opentelemetry/api';
-import { Aggregator } from './export/types';
+import { Accumulator } from './Accumulator';
 
 /**
  * This class represent the base to BoundInstrument, which is responsible for generating
  * the TimeSeries.
  */
 export class BaseBoundInstrument {
-  protected _labels: api.Labels;
-  protected _logger: api.Logger;
-
   constructor(
-    labels: api.Labels,
-    logger: api.Logger,
+    protected readonly _accumulationKey: string,
+    protected readonly _labels: api.Labels,
+    protected readonly _accumulator: Accumulator,
+    protected readonly _logger: api.Logger,
     private readonly _disabled: boolean,
-    private readonly _valueType: api.ValueType,
-    private readonly _aggregator: Aggregator
-  ) {
-    this._labels = labels;
-    this._logger = logger;
-  }
+    private readonly _valueType: api.ValueType
+  ) {}
 
   update(value: number): void {
     if (this._disabled) return;
@@ -56,35 +51,31 @@ export class BaseBoundInstrument {
       value = Math.trunc(value);
     }
 
-    this._aggregator.update(value);
+    this._accumulator.update(this._accumulationKey, this._labels, value);
   }
 
   getLabels(): api.Labels {
     return this._labels;
   }
 
-  getAggregator(): Aggregator {
-    return this._aggregator;
+  getAccumulator(): Accumulator {
+    return this._accumulator;
+  }
+
+  equalWith(other: BaseBoundInstrument): boolean {
+    return (
+      this._accumulator === other._accumulator &&
+      this._accumulationKey === other._accumulationKey
+    );
   }
 }
 
 /**
  * BoundCounter allows the SDK to observe/record a single metric event. The
  * value of single instrument in the `Counter` associated with specified Labels.
  */
-export class BoundCounter
-  extends BaseBoundInstrument
+export class BoundCounter extends BaseBoundInstrument
   implements api.BoundCounter {
-  constructor(
-    labels: api.Labels,
-    disabled: boolean,
-    valueType: api.ValueType,
-    logger: api.Logger,
-    aggregator: Aggregator
-  ) {
-    super(labels, logger, disabled, valueType, aggregator);
-  }
-
   add(value: number): void {
     if (value < 0) {
       this._logger.error(
@@ -102,19 +93,8 @@ export class BoundCounter
  * The value of single instrument in the `UpDownCounter` associated with
  * specified Labels.
  */
-export class BoundUpDownCounter
-  extends BaseBoundInstrument
+export class BoundUpDownCounter extends BaseBoundInstrument
   implements api.BoundCounter {
-  constructor(
-    labels: api.Labels,
-    disabled: boolean,
-    valueType: api.ValueType,
-    logger: api.Logger,
-    aggregator: Aggregator
-  ) {
-    super(labels, logger, disabled, valueType, aggregator);
-  }
-
   add(value: number): void {
     this.update(value);
   }
@@ -123,19 +103,8 @@ export class BoundUpDownCounter
 /**
  * BoundMeasure is an implementation of the {@link BoundMeasure} interface.
  */
-export class BoundValueRecorder
-  extends BaseBoundInstrument
+export class BoundValueRecorder extends BaseBoundInstrument
   implements api.BoundValueRecorder {
-  constructor(
-    labels: api.Labels,
-    disabled: boolean,
-    valueType: api.ValueType,
-    logger: api.Logger,
-    aggregator: Aggregator
-  ) {
-    super(labels, logger, disabled, valueType, aggregator);
-  }
-
   record(value: number): void {
     this.update(value);
   }
@@ -144,16 +113,5 @@ export class BoundValueRecorder
 /**
  * BoundObserver is an implementation of the {@link BoundObserver} interface.
  */
-export class BoundObserver
-  extends BaseBoundInstrument
-  implements api.BoundBaseObserver {
-  constructor(
-    labels: api.Labels,
-    disabled: boolean,
-    valueType: api.ValueType,
-    logger: api.Logger,
-    aggregator: Aggregator
-  ) {
-    super(labels, logger, disabled, valueType, aggregator);
-  }
-}
+export class BoundObserver extends BaseBoundInstrument
+  implements api.BoundBaseObserver {}