feat(aggregators): implement histogram aggregator #927 (#930)

* chore(metrics): move each aggregator in its own file

* feat(aggregators): implement histogram aggregator #927

* chore: address PR comments

* chore: fix ConsoleMetricExporter

* chore: address mayur comments

* chore: add documentation on internal structure

packages/opentelemetry-metrics/src/export/Batcher.ts

@@ -18,7 +18,7 @@ import {
   CounterSumAggregator,
   MeasureExactAggregator,
   ObserverAggregator,
-} from './Aggregator';
+} from './aggregators';
 import { MetricRecord, MetricKind, Aggregator } from './types';
 
 /**

packages/opentelemetry-metrics/src/export/ConsoleMetricExporter.ts

@@ -14,13 +14,7 @@
  * limitations under the License.
  */
 
-import {
-  MetricExporter,
-  MetricRecord,
-  MetricKind,
-  Sum,
-  Distribution,
-} from './types';
+import { MetricExporter, MetricRecord, Distribution, Histogram } from './types';
 import { ExportResult } from '@opentelemetry/base';
 
 /**
@@ -35,25 +29,26 @@ export class ConsoleMetricExporter implements MetricExporter {
     for (const metric of metrics) {
       console.log(metric.descriptor);
       console.log(metric.labels);
-      switch (metric.descriptor.metricKind) {
-        case MetricKind.COUNTER:
-          const sum = metric.aggregator.toPoint().value as Sum;
-          console.log('value: ' + sum);
-          break;
-        default:
-          const distribution = metric.aggregator.toPoint()
-            .value as Distribution;
-          console.log(
-            'min: ' +
-              distribution.min +
-              ', max: ' +
-              distribution.max +
-              ', count: ' +
-              distribution.count +
-              ', sum: ' +
-              distribution.sum
-          );
-          break;
+      const point = metric.aggregator.toPoint();
+      if (typeof point.value === 'number') {
+        console.log('value: ' + point.value);
+      } else if (typeof (point.value as Histogram).buckets === 'object') {
+        const histogram = point.value as Histogram;
+        console.log(
+          `count: ${histogram.count}, sum: ${histogram.sum}, buckets: ${histogram.buckets}`
+        );
+      } else {
+        const distribution = point.value as Distribution;
+        console.log(
+          'min: ' +
+            distribution.min +
+            ', max: ' +
+            distribution.max +
+            ', count: ' +
+            distribution.count +
+            ', sum: ' +
+            distribution.sum
+        );
       }
     }
     return resultCallback(ExportResult.SUCCESS);

packages/opentelemetry-metrics/src/export/aggregators/countersum.ts

@@ -0,0 +1,37 @@
+/*!
+ * Copyright 2020, 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 { Aggregator, Point } from '../types';
+import { HrTime } from '@opentelemetry/api';
+import { hrTime } from '@opentelemetry/core';
+
+/** Basic aggregator which calculates a Sum from individual measurements. */
+export class CounterSumAggregator implements Aggregator {
+  private _current: number = 0;
+  private _lastUpdateTime: HrTime = [0, 0];
+
+  update(value: number): void {
+    this._current += value;
+    this._lastUpdateTime = hrTime();
+  }
+
+  toPoint(): Point {
+    return {
+      value: this._current,
+      timestamp: this._lastUpdateTime,
+    };
+  }
+}

packages/opentelemetry-metrics/src/export/aggregators/histogram.ts

@@ -0,0 +1,81 @@
+/*!
+ * Copyright 2020, 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 { Aggregator, Point, Histogram } from '../types';
+import { HrTime } from '@opentelemetry/api';
+import { hrTime } from '@opentelemetry/core';
+
+/**
+ * Basic aggregator which observes events and counts them in pre-defined buckets
+ * and provides the total sum and count of all observations.
+ */
+export class HistogramAggregator implements Aggregator {
+  private _lastCheckpoint: Histogram;
+  private _currentCheckpoint: Histogram;
+  private _lastCheckpointTime: HrTime;
+  private readonly _boundaries: number[];
+
+  constructor(boundaries: number[]) {
+    if (boundaries === undefined || boundaries.length === 0) {
+      throw new Error(`HistogramAggregator should be created with boundaries.`);
+    }
+    // we need to an ordered set to be able to correctly compute count for each
+    // boundary since we'll iterate on each in order.
+    this._boundaries = boundaries.sort();
+    this._lastCheckpoint = this._newEmptyCheckpoint();
+    this._lastCheckpointTime = hrTime();
+    this._currentCheckpoint = this._newEmptyCheckpoint();
+  }
+
+  update(value: number): void {
+    this._currentCheckpoint.count += 1;
+    this._currentCheckpoint.sum += value;
+
+    for (let i = 0; i < this._boundaries.length; i++) {
+      if (value < this._boundaries[i]) {
+        this._currentCheckpoint.buckets.counts[i] += 1;
+        return;
+      }
+    }
+
+    // value is above all observed boundaries
+    this._currentCheckpoint.buckets.counts[this._boundaries.length] += 1;
+  }
+
+  reset(): void {
+    this._lastCheckpointTime = hrTime();
+    this._lastCheckpoint = this._currentCheckpoint;
+    this._currentCheckpoint = this._newEmptyCheckpoint();
+  }
+
+  toPoint(): Point {
+    return {
+      value: this._lastCheckpoint,
+      timestamp: this._lastCheckpointTime,
+    };
+  }
+
+  private _newEmptyCheckpoint(): Histogram {
+    return {
+      buckets: {
+        boundaries: this._boundaries,
+        counts: this._boundaries.map(() => 0).concat([0]),
+      },
+      sum: 0,
+      count: 0,
+    };
+  }
+}

packages/opentelemetry-metrics/src/export/aggregators/index.ts

@@ -0,0 +1,20 @@
+/*!
+ * Copyright 2020, 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.
+ */
+
+export * from './countersum';
+export * from './observer';
+export * from './measureexact';
+export * from './histogram';

packages/opentelemetry-metrics/src/export/Aggregator.ts → packages/opentelemetry-metrics/src/export/aggregators/measureexact.ts

@@ -14,45 +14,10 @@
  * limitations under the License.
  */
 
-import { Aggregator, Distribution, Point } from './types';
+import { Aggregator, Point } from '../types';
 import { HrTime } from '@opentelemetry/api';
 import { hrTime } from '@opentelemetry/core';
-
-/** Basic aggregator which calculates a Sum from individual measurements. */
-export class CounterSumAggregator implements Aggregator {
-  private _current: number = 0;
-  private _lastUpdateTime: HrTime = [0, 0];
-
-  update(value: number): void {
-    this._current += value;
-    this._lastUpdateTime = hrTime();
-  }
-
-  toPoint(): Point {
-    return {
-      value: this._current,
-      timestamp: this._lastUpdateTime,
-    };
-  }
-}
-
-/** Basic aggregator for Observer which keeps the last recorded value. */
-export class ObserverAggregator implements Aggregator {
-  private _current: number = 0;
-  private _lastUpdateTime: HrTime = [0, 0];
-
-  update(value: number): void {
-    this._current = value;
-    this._lastUpdateTime = hrTime();
-  }
-
-  toPoint(): Point {
-    return {
-      value: this._current,
-      timestamp: this._lastUpdateTime,
-    };
-  }
-}
+import { Distribution } from '../types';
 
 /** Basic aggregator keeping all raw values (events, sum, max and min). */
 export class MeasureExactAggregator implements Aggregator {

packages/opentelemetry-metrics/src/export/aggregators/observer.ts

@@ -0,0 +1,37 @@
+/*!
+ * Copyright 2020, 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 { Aggregator, Point } from '../types';
+import { HrTime } from '@opentelemetry/api';
+import { hrTime } from '@opentelemetry/core';
+
+/** Basic aggregator for Observer which keeps the last recorded value. */
+export class ObserverAggregator implements Aggregator {
+  private _current: number = 0;
+  private _lastUpdateTime: HrTime = [0, 0];
+
+  update(value: number): void {
+    this._current = value;
+    this._lastUpdateTime = hrTime();
+  }
+
+  toPoint(): Point {
+    return {
+      value: this._current,
+      timestamp: this._lastUpdateTime,
+    };
+  }
+}

packages/opentelemetry-metrics/src/export/types.ts

@@ -37,6 +37,33 @@ export interface Distribution {
   sum: number;
 }
 
+export interface Histogram {
+  /**
+   * Buckets are implemented using two different array:
+   *  - boundaries contains every boundary (which are upper boundary for each slice)
+   *  - counts contains count of event for each slice
+   *
+   * Note that we'll always have n+1 (where n is the number of boundaries) slice
+   * because we need to count event that are above the highest boundary. This is the
+   * reason why it's not implement using array of object, because the last slice
+   * dont have any boundary.
+   *
+   * Example if we measure the values: [5, 30, 5, 40, 5, 15, 15, 15, 25]
+   *  with the boundaries [ 10, 20, 30 ], we will have the following state:
+   *
+   * buckets: {
+   *	boundaries: [10, 20, 30],
+   *	counts: [3, 3, 2, 1],
+   * }
+   */
+  buckets: {
+    boundaries: number[];
+    counts: number[];
+  };
+  sum: number;
+  count: number;
+}
+
 export interface MetricRecord {
   readonly descriptor: MetricDescriptor;
   readonly labels: Labels;
@@ -80,6 +107,6 @@ export interface Aggregator {
 }
 
 export interface Point {
-  value: Sum | LastValue | Distribution;
+  value: Sum | LastValue | Distribution | Histogram;
   timestamp: HrTime;
 }

packages/opentelemetry-metrics/src/index.ts

@@ -18,7 +18,6 @@ export * from './BoundInstrument';
 export * from './Meter';
 export * from './Metric';
 export * from './MeterProvider';
-export * from './export/Aggregator';
+export * from './export/aggregators';
 export * from './export/ConsoleMetricExporter';
 export * from './export/types';
-export * from './export/Aggregator';

packages/opentelemetry-metrics/test/Meter.test.ts

@@ -33,7 +33,7 @@ import { NoopLogger, hrTime, hrTimeToNanoseconds } from '@opentelemetry/core';
 import {
   CounterSumAggregator,
   ObserverAggregator,
-} from '../src/export/Aggregator';
+} from '../src/export/aggregators';
 import { ValueType } from '@opentelemetry/api';
 import { Resource } from '@opentelemetry/resources';
 import { hashLabels } from '../src/Utils';