chore: adding metric observable to be able to support async upda… (open-telemetry#964)

* chore: adding metric observable to be able to support async update

* chore: reviews

* chore: reviews

Co-authored-by: Daniel Dyla <dyladan@users.noreply.github.com>

Author: obecny

examples/metrics/metrics/observer.js

@@ -1,6 +1,6 @@
 'use strict';
 
-const { MeterProvider } = require('@opentelemetry/metrics');
+const { MeterProvider, MetricObservable } = require('@opentelemetry/metrics');
 const { PrometheusExporter } = require('@opentelemetry/exporter-prometheus');
 
 const exporter = new PrometheusExporter(
@@ -14,7 +14,7 @@ const exporter = new PrometheusExporter(
 
 const meter = new MeterProvider({
   exporter,
-  interval: 1000,
+  interval: 2000,
 }).getMeter('example-observer');
 
 const otelCpuUsage = meter.createObserver('metric_observer', {
@@ -27,9 +27,16 @@ function getCpuUsage() {
   return Math.random();
 }
 
+const observable = new MetricObservable();
+
+setInterval(() => {
+  observable.next(getCpuUsage());
+}, 5000);
+
 otelCpuUsage.setCallback((observerResult) => {
   observerResult.observe(getCpuUsage, { pid: process.pid, core: '1' });
   observerResult.observe(getCpuUsage, { pid: process.pid, core: '2' });
   observerResult.observe(getCpuUsage, { pid: process.pid, core: '3' });
   observerResult.observe(getCpuUsage, { pid: process.pid, core: '4' });
+  observerResult.observe(observable, { pid: process.pid, core: '5' });
 });

examples/metrics/metrics/observer.png

@@ -26,6 +26,7 @@ export * from './metrics/BoundInstrument';
 export * from './metrics/Meter';
 export * from './metrics/MeterProvider';
 export * from './metrics/Metric';
+export * from './metrics/MetricObservable';
 export * from './metrics/NoopMeter';
 export * from './metrics/NoopMeterProvider';
 export * from './metrics/ObserverResult';

packages/opentelemetry-api/src/index.ts

@@ -0,0 +1,36 @@
+/*!
+ * 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.
+ */
+
+/**
+ * Metric Observable class to handle asynchronous metrics
+ */
+export interface MetricObservable {
+  /**
+   * Sets the next value for observable metric
+   * @param value
+   */
+  next: (value: number) => void;
+  /**
+   * Subscribes for every value change
+   * @param callback
+   */
+  subscribe: (callback: (value: number) => void) => void;
+  /**
+   * Removes the subscriber
+   * @param [callback]
+   */
+  unsubscribe: (callback?: (value: number) => void) => void;
+}

packages/opentelemetry-api/src/metrics/MetricObservable.ts

@@ -15,11 +15,11 @@
  */
 
 import { Labels } from './Metric';
+import { MetricObservable } from './MetricObservable';
 
 /**
  * Interface that is being used in function setCallback for Observer Metric
  */
 export interface ObserverResult {
-  observers: Map<Labels, Function>;
-  observe(callback: Function, labels: Labels): void;
+  observe(callback: Function | MetricObservable, labels: Labels): void;
 }

packages/opentelemetry-api/src/metrics/ObserverResult.ts

@@ -77,7 +77,7 @@
     "ts-node": "^8.6.2",
     "tslint-consistent-codestyle": "^1.16.0",
     "tslint-microsoft-contrib": "^6.2.0",
-    "typescript": "3.6.4",
+    "typescript": "3.7.2",
     "webpack": "^4.35.2",
     "webpack-cli": "^3.3.9",
     "webpack-merge": "^4.2.2"

packages/opentelemetry-exporter-collector/package.json

@@ -25,15 +25,50 @@ const { MeterProvider } = require('@opentelemetry/metrics');
 const meter = new MeterProvider().getMeter('your-meter-name');
 
 const counter = meter.createCounter('metric_name', {
-  labelKeys: ["pid"],
-  description: "Example of a counter"
+  labelKeys: ['pid'],
+  description: 'Example of a counter'
 });
 
 const labels = { pid: process.pid };
 
 // Create a BoundInstrument associated with specified label values.
 const boundCounter = counter.bind(labels);
 boundCounter.add(10);
+
+```
+
+### Observable
+Choose this kind of metric when only last value is important without worry about aggregation
+
+```js
+const { MeterProvider, MetricObservable } = require('@opentelemetry/metrics');
+
+// Initialize the Meter to capture measurements in various ways.
+const meter = new MeterProvider().getMeter('your-meter-name');
+
+const observer = meter.createObserver('metric_name', {
+  labelKeys: ['pid', 'core'],
+  description: 'Example of a observer'
+});
+
+function getCpuUsage() {
+  return Math.random();
+}
+
+const metricObservable = new MetricObservable();
+
+observer.setCallback((observerResult) => {
+  // synchronous callback
+  observerResult.observe(getCpuUsage, { pid: process.pid, core: '1' });
+  // asynchronous callback
+  observerResult.observe(metricObservable, { pid: process.pid, core: '2' });
+});
+
+// simulate asynchronous operation
+setInterval(()=> {
+  metricObservable.next(getCpuUsage());
+}, 2000)
+
 ```
 
 See [examples/prometheus](https://github.com/open-telemetry/opentelemetry-js/tree/master/examples/prometheus) for a short example.

packages/opentelemetry-metrics/README.md

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import * as types from '@opentelemetry/api';
+import * as api from '@opentelemetry/api';
 import { Resource } from '@opentelemetry/resources';
 import {
   BoundCounter,
@@ -30,11 +30,11 @@ import { hashLabels } from './Utils';
 
 /** This is a SDK implementation of {@link Metric} interface. */
 export abstract class Metric<T extends BaseBoundInstrument>
-  implements types.Metric<T> {
+  implements api.Metric<T> {
   protected readonly _monotonic: boolean;
   protected readonly _disabled: boolean;
-  protected readonly _valueType: types.ValueType;
-  protected readonly _logger: types.Logger;
+  protected readonly _valueType: api.ValueType;
+  protected readonly _logger: api.Logger;
   private readonly _descriptor: MetricDescriptor;
   private readonly _instruments: Map<string, T> = new Map();
 
@@ -58,7 +58,7 @@ export abstract class Metric<T extends BaseBoundInstrument>
    * @param labels key-values pairs that are associated with a specific metric
    *     that you want to record.
    */
-  bind(labels: types.Labels): T {
+  bind(labels: api.Labels): T {
     const hash = hashLabels(labels);
     if (this._instruments.has(hash)) return this._instruments.get(hash)!;
 
@@ -71,7 +71,7 @@ export abstract class Metric<T extends BaseBoundInstrument>
    * Removes the Instrument from the metric, if it is present.
    * @param labels key-values pairs that are associated with a specific metric.
    */
-  unbind(labels: types.Labels): void {
+  unbind(labels: api.Labels): void {
     this._instruments.delete(hashLabels(labels));
   }
 
@@ -102,12 +102,12 @@ export abstract class Metric<T extends BaseBoundInstrument>
     };
   }
 
-  protected abstract _makeInstrument(labels: types.Labels): T;
+  protected abstract _makeInstrument(labels: api.Labels): T;
 }
 
 /** This is a SDK implementation of Counter Metric. */
 export class CounterMetric extends Metric<BoundCounter>
-  implements Pick<types.MetricUtils, 'add'> {
+  implements Pick<api.MetricUtils, 'add'> {
   constructor(
     name: string,
     options: MetricOptions,
@@ -116,7 +116,7 @@ export class CounterMetric extends Metric<BoundCounter>
   ) {
     super(name, options, MetricKind.COUNTER, resource);
   }
-  protected _makeInstrument(labels: types.Labels): BoundCounter {
+  protected _makeInstrument(labels: api.Labels): BoundCounter {
     return new BoundCounter(
       labels,
       this._disabled,
@@ -134,13 +134,13 @@ export class CounterMetric extends Metric<BoundCounter>
    * @param labels key-values pairs that are associated with a specific metric
    *     that you want to record.
    */
-  add(value: number, labels: types.Labels) {
+  add(value: number, labels: api.Labels) {
     this.bind(labels).add(value);
   }
 }
 
 export class MeasureMetric extends Metric<BoundMeasure>
-  implements Pick<types.MetricUtils, 'record'> {
+  implements Pick<api.MetricUtils, 'record'> {
   protected readonly _absolute: boolean;
 
   constructor(
@@ -153,7 +153,7 @@ export class MeasureMetric extends Metric<BoundMeasure>
 
     this._absolute = options.absolute !== undefined ? options.absolute : true; // Absolute default is true
   }
-  protected _makeInstrument(labels: types.Labels): BoundMeasure {
+  protected _makeInstrument(labels: api.Labels): BoundMeasure {
     return new BoundMeasure(
       labels,
       this._disabled,
@@ -165,15 +165,15 @@ export class MeasureMetric extends Metric<BoundMeasure>
     );
   }
 
-  record(value: number, labels: types.Labels) {
+  record(value: number, labels: api.Labels) {
     this.bind(labels).record(value);
   }
 }
 
 /** This is a SDK implementation of Observer Metric. */
 export class ObserverMetric extends Metric<BoundObserver>
-  implements Pick<types.MetricUtils, 'setCallback'> {
-  private _observerResult: types.ObserverResult = new ObserverResult();
+  implements Pick<api.MetricUtils, 'setCallback'> {
+  private _observerResult = new ObserverResult();
 
   constructor(
     name: string,
@@ -184,7 +184,7 @@ export class ObserverMetric extends Metric<BoundObserver>
     super(name, options, MetricKind.OBSERVER, resource);
   }
 
-  protected _makeInstrument(labels: types.Labels): BoundObserver {
+  protected _makeInstrument(labels: api.Labels): BoundObserver {
     return new BoundObserver(
       labels,
       this._disabled,
@@ -196,7 +196,7 @@ export class ObserverMetric extends Metric<BoundObserver>
   }
 
   getMetricRecord(): MetricRecord[] {
-    this._observerResult.observers.forEach((callback, labels) => {
+    this._observerResult.callbackObservers.forEach((callback, labels) => {
       const instrument = this.bind(labels);
       instrument.update(callback());
     });
@@ -207,7 +207,13 @@ export class ObserverMetric extends Metric<BoundObserver>
    * Sets a callback where user can observe value for certain labels
    * @param callback
    */
-  setCallback(callback: (observerResult: types.ObserverResult) => void): void {
+  setCallback(callback: (observerResult: api.ObserverResult) => void): void {
     callback(this._observerResult);
+    this._observerResult.observers.forEach((observer, labels) => {
+      observer.subscribe(value => {
+        const instrument = this.bind(labels);
+        instrument.update(value);
+      });
+    });
   }
 }

packages/opentelemetry-metrics/src/Metric.ts

@@ -0,0 +1,51 @@
+/*!
+ * 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 * as api from '@opentelemetry/api';
+
+type Subscriber = (value?: number) => void;
+
+/**
+ * Implements the Metric Observable pattern
+ */
+export class MetricObservable implements api.MetricObservable {
+  private _subscribers: Subscriber[] = [];
+
+  next(value: number) {
+    for (const subscriber of this._subscribers) {
+      subscriber(value);
+    }
+  }
+
+  subscribe(subscriber: Function) {
+    if (typeof subscriber === 'function') {
+      this._subscribers.push(subscriber as Subscriber);
+    }
+  }
+
+  unsubscribe(subscriber?: Function) {
+    if (typeof subscriber === 'function') {
+      for (let i = 0, j = this._subscribers.length; i < j; i++) {
+        if (this._subscribers[i] === subscriber) {
+          this._subscribers.splice(i, 1);
+          break;
+        }
+      }
+    } else {
+      this._subscribers = [];
+    }
+  }
+}

packages/opentelemetry-metrics/src/MetricObservable.ts

@@ -15,6 +15,7 @@
  */
 
 import {
+  MetricObservable,
   ObserverResult as TypeObserverResult,
   Labels,
 } from '@opentelemetry/api';
@@ -23,8 +24,17 @@ import {
  * Implementation of {@link TypeObserverResult}
  */
 export class ObserverResult implements TypeObserverResult {
-  observers = new Map<Labels, Function>();
-  observe(callback: any, labels: Labels): void {
-    this.observers.set(labels, callback);
+  callbackObservers: Map<Labels, Function> = new Map<Labels, Function>();
+  observers: Map<Labels, MetricObservable> = new Map<
+    Labels,
+    MetricObservable
+  >();
+
+  observe(callback: Function | MetricObservable, labels: Labels): void {
+    if (typeof callback === 'function') {
+      this.callbackObservers.set(labels, callback);
+    } else {
+      this.observers.set(labels, callback);
+    }
   }
 }