feat(sdk-metrics-base): shutdown and forceflush on MeterProvider

experimental/packages/opentelemetry-sdk-metrics-base/src/MeterProvider.ts

@@ -18,7 +18,7 @@ import * as api from '@opentelemetry/api';
 import * as metrics from '@opentelemetry/api-metrics';
 import { Resource } from '@opentelemetry/resources';
 import { Meter } from './Meter';
-import { MetricReader } from './export/MetricReader';
+import { MetricReader, ReaderForceFlushOptions, ReaderShutdownOptions } from './export/MetricReader';
 import { MeterProviderSharedState } from './state/MeterProviderSharedState';
 import { InstrumentSelector } from './view/InstrumentSelector';
 import { MeterSelector } from './view/MeterSelector';
@@ -163,59 +163,36 @@ export class MeterProvider implements metrics.MeterProvider {
   /**
    * Flush all buffered data and shut down the MeterProvider and all registered
    * MetricReaders.
-   * Returns a promise which is resolved when all flushes are complete.
    *
-   * TODO: return errors to caller somehow?
+   * Returns a promise which is resolved when all flushes are complete.
    */
-  async shutdown(): Promise<void> {
-    // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#shutdown
-
+  async shutdown(options?: ReaderShutdownOptions): Promise<void> {
     if (this._shutdown) {
       api.diag.warn('shutdown may only be called once per MeterProvider');
       return;
     }
 
-    // TODO add a timeout - spec leaves it up the the SDK if this is configurable
     this._shutdown = true;
 
-    for (const collector of this._sharedState.metricCollectors) {
-      try {
-        await collector.shutdown();
-      } catch (e) {
-        // Log all Errors.
-        if (e instanceof Error) {
-          api.diag.error(`Error shutting down: ${e.message}`);
-        }
-      }
-    }
+    await Promise.all(this._sharedState.metricCollectors.map(collector => {
+      return collector.shutdown(options);
+    }));
   }
 
   /**
    * Notifies all registered MetricReaders to flush any buffered data.
-   * Returns a promise which is resolved when all flushes are complete.
    *
-   * TODO: return errors to caller somehow?
+   * Returns a promise which is resolved when all flushes are complete.
    */
-  async forceFlush(): Promise<void> {
-    // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#forceflush
-
-    // TODO add a timeout - spec leaves it up the the SDK if this is configurable
-
+  async forceFlush(options?: ReaderForceFlushOptions): Promise<void> {
     // do not flush after shutdown
     if (this._shutdown) {
       api.diag.warn('invalid attempt to force flush after shutdown');
       return;
     }
 
-    for (const collector of this._sharedState.metricCollectors) {
-      try {
-        await collector.forceFlush();
-      } catch (e) {
-        // Log all Errors.
-        if (e instanceof Error) {
-          api.diag.error(`Error flushing: ${e.message}`);
-        }
-      }
-    }
+    await Promise.all(this._sharedState.metricCollectors.map(collector => {
+      return collector.forceFlush(options);
+    }));
   }
 }

experimental/packages/opentelemetry-sdk-metrics-base/src/export/MetricReader.ts

@@ -53,6 +53,9 @@ export abstract class MetricReader {
    * @param metricProducer
    */
   setMetricProducer(metricProducer: MetricProducer) {
+    if (this._metricProducer) {
+      throw new Error('MetricReader can not be bound to a MeterProvider again.');
+    }
     this._metricProducer = metricProducer;
     this.onInitialized();
   }

experimental/packages/opentelemetry-sdk-metrics-base/src/state/MetricCollector.ts

@@ -18,7 +18,7 @@ import { hrTime } from '@opentelemetry/core';
 import { AggregationTemporality } from '../export/AggregationTemporality';
 import { ResourceMetrics } from '../export/MetricData';
 import { MetricProducer } from '../export/MetricProducer';
-import { MetricReader } from '../export/MetricReader';
+import { MetricReader, ReaderForceFlushOptions, ReaderShutdownOptions } from '../export/MetricReader';
 import { MeterProviderSharedState } from './MeterProviderSharedState';
 
 /**
@@ -46,15 +46,15 @@ export class MetricCollector implements MetricProducer {
   /**
    * Delegates for MetricReader.forceFlush.
    */
-  async forceFlush(): Promise<void> {
-    await this._metricReader.forceFlush();
+  async forceFlush(options?: ReaderForceFlushOptions): Promise<void> {
+    await this._metricReader.forceFlush(options);
   }
 
   /**
    * Delegates for MetricReader.shutdown.
    */
-  async shutdown(): Promise<void> {
-    await this._metricReader.shutdown();
+  async shutdown(options?: ReaderShutdownOptions): Promise<void> {
+    await this._metricReader.shutdown(options);
   }
 }
 

experimental/packages/opentelemetry-sdk-metrics-base/test/MeterProvider.test.ts

@@ -24,8 +24,13 @@ import {
   defaultResource
 } from './util';
 import { TestMetricReader } from './export/TestMetricReader';
+import * as sinon from 'sinon';
 
 describe('MeterProvider', () => {
+  afterEach(() => {
+    sinon.restore();
+  });
+
   describe('constructor', () => {
     it('should construct without exceptions', () => {
       const meterProvider = new MeterProvider();
@@ -422,4 +427,47 @@ describe('MeterProvider', () => {
       });
     });
   });
+
+  describe('shutdown', () => {
+    it('should shutdown all registered metric readers', async () => {
+      const meterProvider = new MeterProvider({ resource: defaultResource });
+      const reader1 = new TestMetricReader();
+      const reader2 = new TestMetricReader();
+      const reader1ShutdownSpy = sinon.spy(reader1, 'shutdown');
+      const reader2ShutdownSpy = sinon.spy(reader2, 'shutdown');
+
+      meterProvider.addMetricReader(reader1);
+      meterProvider.addMetricReader(reader2);
+
+      await meterProvider.shutdown();
+      await meterProvider.shutdown();
+      await meterProvider.shutdown();
+
+      assert.strictEqual(reader1ShutdownSpy.callCount, 1);
+      assert.strictEqual(reader2ShutdownSpy.callCount, 1);
+    });
+  });
+
+  describe('forceFlush', () => {
+    it('should forceFlush all registered metric readers', async () => {
+      const meterProvider = new MeterProvider({ resource: defaultResource });
+      const reader1 = new TestMetricReader();
+      const reader2 = new TestMetricReader();
+      const reader1ForceFlushSpy = sinon.spy(reader1, 'forceFlush');
+      const reader2ForceFlushSpy = sinon.spy(reader2, 'forceFlush');
+
+      meterProvider.addMetricReader(reader1);
+      meterProvider.addMetricReader(reader2);
+
+      await meterProvider.forceFlush();
+      await meterProvider.forceFlush();
+      assert.strictEqual(reader1ForceFlushSpy.callCount, 2);
+      assert.strictEqual(reader2ForceFlushSpy.callCount, 2);
+
+      await meterProvider.shutdown();
+      await meterProvider.forceFlush();
+      assert.strictEqual(reader1ForceFlushSpy.callCount, 2);
+      assert.strictEqual(reader2ForceFlushSpy.callCount, 2);
+    });
+  });
 });

experimental/packages/opentelemetry-sdk-metrics-base/test/export/MetricReader.test.ts

@@ -0,0 +1,34 @@
+/*
+ * 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 * as assert from 'assert';
+import { MeterProvider } from '../../src/MeterProvider';
+import { TestMetricReader } from './TestMetricReader';
+
+
+describe('MetricReader', () => {
+  describe('setMetricProducer', () => {
+    it('The SDK MUST NOT allow a MetricReader instance to be registered on more than one MeterProvider instance', () => {
+      const reader = new TestMetricReader();
+      const meterProvider1 = new MeterProvider();
+      const meterProvider2 = new MeterProvider();
+
+      meterProvider1.addMetricReader(reader);
+      assert.throws(() => meterProvider1.addMetricReader(reader), /MetricReader can not be bound to a MeterProvider again/);
+      assert.throws(() => meterProvider2.addMetricReader(reader), /MetricReader can not be bound to a MeterProvider again/);
+    });
+  });
+});