Batch observer (open-telemetry#1137)

* chore: adding batch observer, some metrics refactoring

* chore: undo changes after testing

* chore: undo changes after testing

* chore: addressing comments

* chore: renaming observer into value observer, fixing few spotted issues

* chore: missing renamed for ValueObserver

* chore: removing unused class

* chore: cleanup

* chore: refactoring, renaming aggregators

* chore: refactoring observer to have base class that can be extended

* chore: changing aggregator for ValueObserver, exposing batcher so it can be used to override a default one

* chore: addressing comments

* chore: addressing comments

* chore: preventing user from updating observer after timeout or update

* chore: aligning aggregators for value observer and recorder with regards to last spec changes

* chore: fixing test

* chore: fixes after merge

* chore: changes after review

* chore: changes after review with some additional fixes around typing

* chore: changes after review

* chore: lint

* chore: reviews

* chore: typo

api/src/index.ts

@@ -22,13 +22,14 @@ export * from './context/propagation/NoopHttpTextPropagator';
 export * from './context/propagation/setter';
 export * from './correlation_context/CorrelationContext';
 export * from './correlation_context/EntryValue';
+export * from './metrics/BatchObserverResult';
 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/Observation';
 export * from './metrics/ObserverResult';
 export * from './trace/attributes';
 export * from './trace/Event';

api/src/metrics/MetricObservable.ts → api/src/metrics/BatchObserverResult.ts

@@ -13,20 +13,19 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-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;
+
+import { Labels } from './Metric';
+import { Observation } from './Observation';
+
+/**
+ * Interface that is being used in callback function for Observer Metric
+ * for batch
+ */
+export interface BatchObserverResult {
   /**
-   * Removes the subscriber
-   * @param [callback]
+   * Used to observe (update) observations for certain labels
+   * @param labels
+   * @param observations
    */
-  unsubscribe: (callback?: (value: number) => void) => void;
+  observe(labels: Labels, observations: Observation[]): void;
 }

api/src/metrics/BoundInstrument.ts

@@ -44,3 +44,8 @@ export interface BoundValueRecorder {
     spanContext: SpanContext
   ): void;
 }
+
+/** An Instrument for Base Observer */
+export interface BoundBaseObserver {
+  update(value: number): void;
+}

api/src/metrics/Meter.ts

@@ -14,13 +14,17 @@
  * limitations under the License.
  */
 
+import { BatchObserverResult } from './BatchObserverResult';
 import {
   MetricOptions,
   Counter,
   ValueRecorder,
-  Observer,
+  ValueObserver,
+  BatchObserver,
+  BatchMetricOptions,
   UpDownCounter,
 } from './Metric';
+import { ObserverResult } from './ObserverResult';
 
 /**
  * An interface to allow the recording metrics.
@@ -66,9 +70,27 @@ export interface Meter {
   createUpDownCounter(name: string, options?: MetricOptions): UpDownCounter;
 
   /**
-   * Creates a new `Observer` metric.
+   * Creates a new `ValueObserver` metric.
    * @param name the name of the metric.
    * @param [options] the metric options.
+   * @param [callback] the observer callback
    */
-  createObserver(name: string, options?: MetricOptions): Observer;
+  createValueObserver(
+    name: string,
+    options?: MetricOptions,
+    callback?: (observerResult: ObserverResult) => void
+  ): ValueObserver;
+
+  /**
+   * Creates a new `BatchObserver` metric, can be used to update many metrics
+   * at the same time and when operations needs to be async
+   * @param name the name of the metric.
+   * @param callback the batch observer callback
+   * @param [options] the metric batch options.
+   */
+  createBatchObserver(
+    name: string,
+    callback: (batchObserverResult: BatchObserverResult) => void,
+    options?: BatchMetricOptions
+  ): BatchObserver;
 }

api/src/metrics/Metric.ts

@@ -16,8 +16,12 @@
 
 import { CorrelationContext } from '../correlation_context/CorrelationContext';
 import { SpanContext } from '../trace/span_context';
-import { ObserverResult } from './ObserverResult';
-import { BoundCounter, BoundValueRecorder } from './BoundInstrument';
+import {
+  BoundBaseObserver,
+  BoundCounter,
+  BoundValueRecorder,
+} from './BoundInstrument';
+import { Logger } from '../common/Logger';
 
 /**
  * Options needed for metric creation
@@ -58,6 +62,18 @@ export interface MetricOptions {
    * @default {@link ValueType.DOUBLE}
    */
   valueType?: ValueType;
+
+  /**
+   * User provided logger.
+   */
+  logger?: Logger;
+}
+
+export interface BatchMetricOptions extends MetricOptions {
+  /**
+   * Indicates how long the batch metric should wait to update before cancel
+   */
+  maxTimeoutUpdateMS?: number;
 }
 
 /** The Type of value. It describes how the data is reported. */
@@ -148,16 +164,21 @@ export interface ValueRecorder extends UnboundMetric<BoundValueRecorder> {
 }
 
 /** Base interface for the Observer metrics. */
-export interface Observer extends Metric {
-  /**
-   * Sets a callback where user can observe value for certain labels. The
-   * observers are called periodically to retrieve the value.
-   * @param callback a function that will be called once to set observers
-   *     for values
-   */
-  setCallback(callback: (observerResult: ObserverResult) => void): void;
+export interface BaseObserver extends UnboundMetric<BoundBaseObserver> {
+  observation: (
+    value: number
+  ) => {
+    value: number;
+    observer: BaseObserver;
+  };
 }
 
+/** Base interface for the Value Observer metrics. */
+export type ValueObserver = BaseObserver;
+
+/** Base interface for the Batch Observer metrics. */
+export type BatchObserver = Metric;
+
 /**
  * key-value pairs passed by the user.
  */

api/src/metrics/NoopMeter.ts

@@ -14,17 +14,24 @@
  * limitations under the License.
  */
 
+import { BatchObserverResult } from './BatchObserverResult';
 import { Meter } from './Meter';
 import {
   MetricOptions,
   UnboundMetric,
   Labels,
   Counter,
   ValueRecorder,
-  Observer,
+  ValueObserver,
+  BatchObserver,
   UpDownCounter,
+  BaseObserver,
 } from './Metric';
-import { BoundValueRecorder, BoundCounter } from './BoundInstrument';
+import {
+  BoundValueRecorder,
+  BoundCounter,
+  BoundBaseObserver,
+} from './BoundInstrument';
 import { CorrelationContext } from '../correlation_context/CorrelationContext';
 import { SpanContext } from '../trace/span_context';
 import { ObserverResult } from './ObserverResult';
@@ -64,12 +71,29 @@ export class NoopMeter implements Meter {
   }
 
   /**
-   * Returns constant noop observer.
+   * Returns constant noop value observer.
    * @param name the name of the metric.
    * @param [options] the metric options.
+   * @param [callback] the value observer callback
+   */
+  createValueObserver(
+    name: string,
+    options?: MetricOptions,
+    callback?: (observerResult: ObserverResult) => void
+  ): ValueObserver {
+    return NOOP_VALUE_OBSERVER_METRIC;
+  }
+
+  /**
+   * Returns constant noop batch observer.
+   * @param name the name of the metric.
+   * @param callback the batch observer callback
    */
-  createObserver(name: string, options?: MetricOptions): Observer {
-    return NOOP_OBSERVER_METRIC;
+  createBatchObserver(
+    name: string,
+    callback: (batchObserverResult: BatchObserverResult) => void
+  ): BatchObserver {
+    return NOOP_BATCH_OBSERVER_METRIC;
   }
 }
 
@@ -79,6 +103,7 @@ export class NoopMetric<T> implements UnboundMetric<T> {
   constructor(instrument: T) {
     this._instrument = instrument;
   }
+
   /**
    * Returns a Bound Instrument associated with specified Labels.
    * It is recommended to keep a reference to the Bound Instrument instead of
@@ -131,10 +156,19 @@ export class NoopValueRecorderMetric extends NoopMetric<BoundValueRecorder>
   }
 }
 
-export class NoopObserverMetric extends NoopMetric<void> implements Observer {
-  setCallback(callback: (observerResult: ObserverResult) => void): void {}
+export class NoopBaseObserverMetric extends NoopMetric<BoundBaseObserver>
+  implements BaseObserver {
+  observation() {
+    return {
+      observer: this as BaseObserver,
+      value: 0,
+    };
+  }
 }
 
+export class NoopBatchObserverMetric extends NoopMetric<void>
+  implements BatchObserver {}
+
 export class NoopBoundCounter implements BoundCounter {
   add(value: number): void {
     return;
@@ -151,6 +185,10 @@ export class NoopBoundValueRecorder implements BoundValueRecorder {
   }
 }
 
+export class NoopBoundBaseObserver implements BoundBaseObserver {
+  update(value: number) {}
+}
+
 export const NOOP_METER = new NoopMeter();
 export const NOOP_BOUND_COUNTER = new NoopBoundCounter();
 export const NOOP_COUNTER_METRIC = new NoopCounterMetric(NOOP_BOUND_COUNTER);
@@ -160,4 +198,8 @@ export const NOOP_VALUE_RECORDER_METRIC = new NoopValueRecorderMetric(
   NOOP_BOUND_VALUE_RECORDER
 );
 
-export const NOOP_OBSERVER_METRIC = new NoopObserverMetric();
+export const NOOP_BOUND_BASE_OBSERVER = new NoopBoundBaseObserver();
+export const NOOP_VALUE_OBSERVER_METRIC = new NoopBaseObserverMetric(
+  NOOP_BOUND_BASE_OBSERVER
+);
+export const NOOP_BATCH_OBSERVER_METRIC = new NoopBatchObserverMetric();

api/src/metrics/Observation.ts

@@ -0,0 +1,25 @@
+/*
+ * 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 { BaseObserver } from './Metric';
+
+/**
+ * Interface for updating value of certain value observer
+ */
+export interface Observation {
+  observer: BaseObserver;
+  value: number;
+}

api/src/metrics/ObserverResult.ts

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