feat(meter): allow custom batcher #932 (#933)

Co-authored-by: Mayur Kale <[email protected]>

Author: vmarchaud

packages/opentelemetry-metrics/src/Meter.ts

@@ -43,7 +43,7 @@ export class Meter implements types.Meter {
    */
   constructor(config: MeterConfig = DEFAULT_CONFIG) {
     this._logger = config.logger || new ConsoleLogger(config.logLevel);
-    this._batcher = new UngroupedBatcher();
+    this._batcher = config.batcher ?? new UngroupedBatcher();
     this._resource = config.resource || Resource.createTelemetrySDKResource();
     // start the push controller
     const exporter = config.exporter || new NoopExporter();

packages/opentelemetry-metrics/src/types.ts

@@ -18,6 +18,7 @@ import { LogLevel } from '@opentelemetry/core';
 import { Logger, ValueType } from '@opentelemetry/api';
 import { MetricExporter } from './export/types';
 import { Resource } from '@opentelemetry/resources';
+import { Batcher } from './export/Batcher';
 
 /** Options needed for SDK metric creation. */
 export interface MetricOptions {
@@ -68,6 +69,9 @@ export interface MeterConfig {
 
   /** Resource associated with metric telemetry */
   resource?: Resource;
+
+  /** Metric batcher. */
+  batcher?: Batcher;
 }
 
 /** Default Meter configuration. */

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

@@ -26,6 +26,7 @@ import {
   Distribution,
   ObserverMetric,
   MetricRecord,
+  Aggregator,
 } from '../src';
 import * as types from '@opentelemetry/api';
 import { NoopLogger, hrTime, hrTimeToNanoseconds } from '@opentelemetry/core';
@@ -36,6 +37,7 @@ import {
 import { ValueType } from '@opentelemetry/api';
 import { Resource } from '@opentelemetry/resources';
 import { hashLabels } from '../src/Utils';
+import { Batcher } from '../src/export/Batcher';
 
 describe('Meter', () => {
   let meter: Meter;
@@ -538,8 +540,27 @@ describe('Meter', () => {
       assert.strictEqual(value, 10);
     });
   });
+
+  it('should allow custom batcher', () => {
+    const customMeter = new MeterProvider().getMeter('custom-batcher', '*', {
+      batcher: new CustomBatcher(),
+    });
+    assert.throws(() => {
+      const measure = customMeter.createMeasure('myMeasure');
+      measure.bind({}).record(1);
+    }, /aggregatorFor method not implemented/);
+  });
 });
 
+class CustomBatcher extends Batcher {
+  process(record: MetricRecord): void {
+    throw new Error('process method not implemented.');
+  }
+  aggregatorFor(metricKind: MetricKind): Aggregator {
+    throw new Error('aggregatorFor method not implemented.');
+  }
+}
+
 function ensureMetric(metric: MetricRecord) {
   assert.ok(metric.aggregator instanceof ObserverAggregator);
   assert.ok(