Metrics: Add lastUpdateTimestamp associated with point (#893)

* Metrics: Add lastUpdateTimestamp associated with point

* update lastUpdateTime in MeasureExactAggregator

* add tests

Co-authored-by: Daniel Dyla <[email protected]>

Author: mayurkale22

packages/opentelemetry-exporter-prometheus/src/prometheus.ts

@@ -15,7 +15,7 @@
  */
 
 import { ExportResult } from '@opentelemetry/base';
-import { NoopLogger } from '@opentelemetry/core';
+import { NoopLogger, hrTimeToMilliseconds } from '@opentelemetry/core';
 import {
   CounterSumAggregator,
   LastValue,
@@ -124,27 +124,30 @@ export class PrometheusExporter implements MetricExporter {
     if (!metric) return;
 
     const labelKeys = record.descriptor.labelKeys;
-    const value = record.aggregator.value();
+    const point = record.aggregator.toPoint();
 
     if (metric instanceof Counter) {
       // Prometheus counter saves internal state and increments by given value.
       // MetricRecord value is the current state, not the delta to be incremented by.
       // Currently, _registerMetric creates a new counter every time the value changes,
       // so the increment here behaves as a set value (increment from 0)
-      metric.inc(this._getLabelValues(labelKeys, record.labels), value as Sum);
+      metric.inc(
+        this._getLabelValues(labelKeys, record.labels),
+        point.value as Sum
+      );
     }
 
     if (metric instanceof Gauge) {
       if (record.aggregator instanceof CounterSumAggregator) {
         metric.set(
           this._getLabelValues(labelKeys, record.labels),
-          value as Sum
+          point.value as Sum
         );
       } else if (record.aggregator instanceof ObserverAggregator) {
         metric.set(
           this._getLabelValues(labelKeys, record.labels),
-          value as LastValue,
-          new Date()
+          point.value as LastValue,
+          hrTimeToMilliseconds(point.timestamp)
         );
       }
     }

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

@@ -14,37 +14,50 @@
  * limitations under the License.
  */
 
-import { Aggregator, Distribution, LastValue, Sum } from './types';
+import { Aggregator, Distribution, 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();
   }
 
-  value(): Sum {
-    return this._current;
+  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();
   }
 
-  value(): LastValue {
-    return this._current;
+  toPoint(): Point {
+    return {
+      value: this._current,
+      timestamp: this._lastUpdateTime,
+    };
   }
 }
 
 /** Basic aggregator keeping all raw values (events, sum, max and min). */
 export class MeasureExactAggregator implements Aggregator {
   private _distribution: Distribution;
+  private _lastUpdateTime: HrTime = [0, 0];
 
   constructor() {
     this._distribution = {
@@ -60,9 +73,13 @@ export class MeasureExactAggregator implements Aggregator {
     this._distribution.sum += value;
     this._distribution.min = Math.min(this._distribution.min, value);
     this._distribution.max = Math.max(this._distribution.max, value);
+    this._lastUpdateTime = hrTime();
   }
 
-  value(): Distribution {
-    return this._distribution;
+  toPoint(): Point {
+    return {
+      value: this._distribution,
+      timestamp: this._lastUpdateTime,
+    };
   }
 }

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

@@ -37,11 +37,12 @@ export class ConsoleMetricExporter implements MetricExporter {
       console.log(metric.labels.labels);
       switch (metric.descriptor.metricKind) {
         case MetricKind.COUNTER:
-          const sum = metric.aggregator.value() as Sum;
+          const sum = metric.aggregator.toPoint().value as Sum;
           console.log('value: ' + sum);
           break;
         default:
-          const distribution = metric.aggregator.value() as Distribution;
+          const distribution = metric.aggregator.toPoint()
+            .value as Distribution;
           console.log(
             'min: ' +
               distribution.min +

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import { ValueType } from '@opentelemetry/api';
+import { ValueType, HrTime } from '@opentelemetry/api';
 import { ExportResult } from '@opentelemetry/base';
 import { LabelSet } from '../LabelSet';
 
@@ -76,6 +76,11 @@ export interface Aggregator {
   /** Updates the current with the new value. */
   update(value: number): void;
 
-  /** Returns snapshot of the current value. */
-  value(): Sum | Distribution;
+  /** Returns snapshot of the current point (value with timestamp). */
+  toPoint(): Point;
+}
+
+export interface Point {
+  value: Sum | LastValue | Distribution;
+  timestamp: HrTime;
 }

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

@@ -29,7 +29,7 @@ import {
 } from '../src';
 import * as types from '@opentelemetry/api';
 import { LabelSet } from '../src/LabelSet';
-import { NoopLogger } from '@opentelemetry/core';
+import { NoopLogger, hrTime, hrTimeToNanoseconds } from '@opentelemetry/core';
 import {
   CounterSumAggregator,
   ObserverAggregator,
@@ -63,6 +63,8 @@ describe('Meter', () => {
   });
 
   describe('#counter', () => {
+    const performanceTimeOrigin = hrTime();
+
     it('should create a counter', () => {
       const counter = meter.createCounter('name');
       assert.ok(counter instanceof Metric);
@@ -84,9 +86,19 @@ describe('Meter', () => {
       meter.collect();
       const [record1] = meter.getBatcher().checkPointSet();
 
-      assert.strictEqual(record1.aggregator.value(), 10);
+      assert.strictEqual(record1.aggregator.toPoint().value, 10);
+      const lastTimestamp = record1.aggregator.toPoint().timestamp;
+      assert.ok(
+        hrTimeToNanoseconds(lastTimestamp) >
+          hrTimeToNanoseconds(performanceTimeOrigin)
+      );
       counter.add(10, labelSet);
-      assert.strictEqual(record1.aggregator.value(), 20);
+      assert.strictEqual(record1.aggregator.toPoint().value, 20);
+
+      assert.ok(
+        hrTimeToNanoseconds(record1.aggregator.toPoint().timestamp) >
+          hrTimeToNanoseconds(lastTimestamp)
+      );
     });
 
     it('should return counter with resource', () => {
@@ -102,9 +114,9 @@ describe('Meter', () => {
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
 
-        assert.strictEqual(record1.aggregator.value(), 10);
+        assert.strictEqual(record1.aggregator.toPoint().value, 10);
         boundCounter.add(10);
-        assert.strictEqual(record1.aggregator.value(), 20);
+        assert.strictEqual(record1.aggregator.toPoint().value, 20);
       });
 
       it('should return the aggregator', () => {
@@ -123,9 +135,9 @@ describe('Meter', () => {
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
 
-        assert.strictEqual(record1.aggregator.value(), 10);
+        assert.strictEqual(record1.aggregator.toPoint().value, 10);
         boundCounter.add(-100);
-        assert.strictEqual(record1.aggregator.value(), 10);
+        assert.strictEqual(record1.aggregator.toPoint().value, 10);
       });
 
       it('should not add the instrument data when disabled', () => {
@@ -136,7 +148,7 @@ describe('Meter', () => {
         boundCounter.add(10);
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.strictEqual(record1.aggregator.value(), 0);
+        assert.strictEqual(record1.aggregator.toPoint().value, 0);
       });
 
       it('should add negative value when monotonic is set to false', () => {
@@ -147,7 +159,7 @@ describe('Meter', () => {
         boundCounter.add(-10);
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.strictEqual(record1.aggregator.value(), -10);
+        assert.strictEqual(record1.aggregator.toPoint().value, -10);
       });
 
       it('should return same instrument on same label values', () => {
@@ -159,7 +171,7 @@ describe('Meter', () => {
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
 
-        assert.strictEqual(record1.aggregator.value(), 20);
+        assert.strictEqual(record1.aggregator.toPoint().value, 20);
         assert.strictEqual(boundCounter, boundCounter1);
       });
     });
@@ -214,7 +226,7 @@ describe('Meter', () => {
           unit: '1',
           valueType: ValueType.DOUBLE,
         });
-        assert.strictEqual(record[0].aggregator.value(), 10);
+        assert.strictEqual(record[0].aggregator.toPoint().value, 10);
       });
     });
 
@@ -306,6 +318,8 @@ describe('Meter', () => {
     });
 
     describe('.bind()', () => {
+      const performanceTimeOrigin = hrTime();
+
       it('should create a measure instrument', () => {
         const measure = meter.createMeasure('name') as MeasureMetric;
         const boundMeasure = measure.bind(labelSet);
@@ -319,12 +333,15 @@ describe('Meter', () => {
 
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.deepStrictEqual(record1.aggregator.value() as Distribution, {
-          count: 0,
-          max: -Infinity,
-          min: Infinity,
-          sum: 0,
-        });
+        assert.deepStrictEqual(
+          record1.aggregator.toPoint().value as Distribution,
+          {
+            count: 0,
+            max: -Infinity,
+            min: Infinity,
+            sum: 0,
+          }
+        );
       });
 
       it('should not set the instrument data when disabled', () => {
@@ -336,12 +353,15 @@ describe('Meter', () => {
 
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.deepStrictEqual(record1.aggregator.value() as Distribution, {
-          count: 0,
-          max: -Infinity,
-          min: Infinity,
-          sum: 0,
-        });
+        assert.deepStrictEqual(
+          record1.aggregator.toPoint().value as Distribution,
+          {
+            count: 0,
+            max: -Infinity,
+            min: Infinity,
+            sum: 0,
+          }
+        );
       });
 
       it('should accept negative (and positive) values when absolute is set to false', () => {
@@ -354,12 +374,19 @@ describe('Meter', () => {
 
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.deepStrictEqual(record1.aggregator.value() as Distribution, {
-          count: 2,
-          max: 50,
-          min: -10,
-          sum: 40,
-        });
+        assert.deepStrictEqual(
+          record1.aggregator.toPoint().value as Distribution,
+          {
+            count: 2,
+            max: 50,
+            min: -10,
+            sum: 40,
+          }
+        );
+        assert.ok(
+          hrTimeToNanoseconds(record1.aggregator.toPoint().timestamp) >
+            hrTimeToNanoseconds(performanceTimeOrigin)
+        );
       });
 
       it('should return same instrument on same label values', () => {
@@ -370,12 +397,15 @@ describe('Meter', () => {
         boundMeasure2.record(100);
         meter.collect();
         const [record1] = meter.getBatcher().checkPointSet();
-        assert.deepStrictEqual(record1.aggregator.value() as Distribution, {
-          count: 2,
-          max: 100,
-          min: 10,
-          sum: 110,
-        });
+        assert.deepStrictEqual(
+          record1.aggregator.toPoint().value as Distribution,
+          {
+            count: 2,
+            max: 100,
+            min: 10,
+            sum: 110,
+          }
+        );
         assert.strictEqual(boundMeasure1, boundMeasure2);
       });
     });
@@ -500,7 +530,7 @@ describe('Meter', () => {
         labelKeys: ['key'],
       });
       assert.strictEqual(record[0].labels, labelSet);
-      const value = record[0].aggregator.value() as Sum;
+      const value = record[0].aggregator.toPoint().value as Sum;
       assert.strictEqual(value, 10.45);
     });
 
@@ -529,16 +559,22 @@ describe('Meter', () => {
         labelKeys: ['key'],
       });
       assert.strictEqual(record[0].labels, labelSet);
-      const value = record[0].aggregator.value() as Sum;
+      const value = record[0].aggregator.toPoint().value as Sum;
       assert.strictEqual(value, 10);
     });
   });
 });
 
 function ensureMetric(metric: MetricRecord) {
   assert.ok(metric.aggregator instanceof ObserverAggregator);
-  assert.ok(metric.aggregator.value() >= 0 && metric.aggregator.value() <= 1);
-  assert.ok(metric.aggregator.value() >= 0 && metric.aggregator.value() <= 1);
+  assert.ok(
+    metric.aggregator.toPoint().value >= 0 &&
+      metric.aggregator.toPoint().value <= 1
+  );
+  assert.ok(
+    metric.aggregator.toPoint().value >= 0 &&
+      metric.aggregator.toPoint().value <= 1
+  );
   const descriptor = metric.descriptor;
   assert.strictEqual(descriptor.name, 'name');
   assert.strictEqual(descriptor.description, 'desc');