feat(sdk-metrics): bootstrap metric streams

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

@@ -16,6 +16,8 @@
 
 import { MetricOptions, ValueType } from '@opentelemetry/api-metrics-wip';
 import { InstrumentType } from './Instruments';
+import { View } from './view/View';
+
 
 export interface InstrumentDescriptor {
   readonly name: string;
@@ -34,3 +36,13 @@ export function createInstrumentDescriptor(name: string, type: InstrumentType, o
     valueType: options?.valueType ?? ValueType.DOUBLE,
   };
 }
+
+export function createInstrumentDescriptorWithView(view: View, instrument: InstrumentDescriptor): InstrumentDescriptor {
+  return {
+    name: view.name ?? instrument.name,
+    description: view.description ?? instrument.description,
+    type: instrument.type,
+    unit: instrument.unit,
+    valueType: instrument.valueType,
+  };
+}

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

@@ -20,12 +20,17 @@ import { createInstrumentDescriptor, InstrumentDescriptor } from './InstrumentDe
 import { Counter, Histogram, InstrumentType, UpDownCounter } from './Instruments';
 import { MeterProviderSharedState } from './state/MeterProviderSharedState';
 import { MultiMetricStorage } from './state/MultiWritableMetricStorage';
-import { NoopWritableMetricStorage, WritableMetricStorage } from './state/WritableMetricStorage';
+import { SyncMetricStorage } from './state/SyncMetricStorage';
+import { MetricStorage } from './state/MetricStorage';
+import { MetricData } from './export/MetricData';
+import { isNotNullish } from './utils';
+import { MetricCollectorHandle } from './state/MetricCollector';
+import { HrTime } from '@opentelemetry/api';
 
 // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#meter
 
 export class Meter implements metrics.Meter {
-  private _metricStorageRegistry = new Map<string, WritableMetricStorage>();
+  private _metricStorageRegistry = new Map<string, MetricStorage>();
 
   // instrumentation library required by spec to be on meter
   // spec requires provider config changes to apply to previously created meters, achieved by holding a reference to the provider
@@ -80,9 +85,8 @@ export class Meter implements metrics.Meter {
 
   private _registerMetricStorage(descriptor: InstrumentDescriptor) {
     const views = this._meterProviderSharedState.viewRegistry.findViews(descriptor, this._instrumentationLibrary);
-    const storages = views.map(_view => {
-      // TODO: create actual metric storages.
-      const storage = new NoopWritableMetricStorage();
+    const storages = views.map(view => {
+      const storage = SyncMetricStorage.create(view, descriptor);
       // TODO: handle conflicts
       this._metricStorageRegistry.set(descriptor.name, storage);
       return storage;
@@ -92,4 +96,17 @@ export class Meter implements metrics.Meter {
     }
     return new MultiMetricStorage(storages);
   }
+
+  async collect(collector: MetricCollectorHandle, collectionTime: HrTime): Promise<MetricData[]> {
+    const result = await Promise.all(Array.from(this._metricStorageRegistry.values()).map(metricStorage => {
+      return metricStorage.collect(
+        collector,
+        this._meterProviderSharedState.metricCollectors,
+        this._meterProviderSharedState.resource,
+        this._instrumentationLibrary,
+        this._meterProviderSharedState.sdkStartTime,
+        collectionTime);
+    }));
+    return result.filter(isNotNullish);
+  }
 }

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

@@ -18,13 +18,12 @@ import * as api from '@opentelemetry/api';
 import * as metrics from '@opentelemetry/api-metrics-wip';
 import { Resource } from '@opentelemetry/resources';
 import { Meter } from './Meter';
-import { MetricExporter } from './MetricExporter';
-import { MetricReader } from './MetricReader';
+import { MetricReader } from './export/MetricReader';
 import { MeterProviderSharedState } from './state/MeterProviderSharedState';
 import { InstrumentSelector } from './view/InstrumentSelector';
 import { MeterSelector } from './view/MeterSelector';
 import { View } from './view/View';
-
+import { MetricCollector } from './state/MetricCollector';
 
 // https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#meterprovider
 
@@ -35,8 +34,6 @@ export type MeterProviderOptions = {
 export class MeterProvider {
   private _sharedState: MeterProviderSharedState;
   private _shutdown = false;
-  private _metricReaders: MetricReader[] = [];
-  private _metricExporters: MetricExporter[] = [];
 
   constructor(options: MeterProviderOptions) {
     this._sharedState = new MeterProviderSharedState(options.resource ?? Resource.empty());
@@ -57,7 +54,9 @@ export class MeterProvider {
   }
 
   addMetricReader(metricReader: MetricReader) {
-    this._metricReaders.push(metricReader);
+    const collector = new MetricCollector(this._sharedState, metricReader);
+    metricReader.setMetricProducer(collector);
+    this._sharedState.metricCollectors.push(collector);
   }
 
   addView(view: View, instrumentSelector: InstrumentSelector, meterSelector: MeterSelector) {
@@ -66,7 +65,8 @@ export class MeterProvider {
   }
 
   /**
-   * Flush all buffered data and shut down the MeterProvider and all exporters and metric readers.
+   * 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?
@@ -82,12 +82,11 @@ export class MeterProvider {
     // TODO add a timeout - spec leaves it up the the SDK if this is configurable
     this._shutdown = true;
 
-    // Shut down all exporters and readers.
-    // Log all Errors.
-    for (const exporter of this._metricExporters) {
+    for (const collector of this._sharedState.metricCollectors) {
       try {
-        await exporter.shutdown();
+        await collector.shutdown();
       } catch (e) {
+        // Log all Errors.
         if (e instanceof Error) {
           api.diag.error(`Error shutting down: ${e.message}`)
         }
@@ -96,7 +95,7 @@ export class MeterProvider {
   }
 
   /**
-   * Notifies all exporters and metric readers to flush any buffered data.
+   * 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?
@@ -112,10 +111,11 @@ export class MeterProvider {
       return;
     }
 
-    for (const exporter of [...this._metricExporters, ...this._metricReaders]) {
+    for (const collector of this._sharedState.metricCollectors) {
       try {
-        await exporter.forceFlush();
+        await collector.forceFlush();
       } catch (e) {
+        // Log all Errors.
         if (e instanceof Error) {
           api.diag.error(`Error flushing: ${e.message}`)
         }

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

@@ -0,0 +1,61 @@
+/*
+ * 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 { AggregationTemporality } from './AggregationTemporality';
+import { MetricData } from './MetricData';
+
+
+// https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#metricexporter
+
+// TODO should this just be an interface and exporters can implement their own shutdown?
+export abstract class MetricExporter {
+  protected _shutdown = false;
+
+  abstract export(batch: MetricData[]): Promise<void>;
+
+  abstract forceFlush(): Promise<void>;
+
+  abstract getPreferredAggregationTemporality(): AggregationTemporality;
+
+  async shutdown(): Promise<void> {
+    if (this._shutdown) {
+      return;
+    }
+
+    // Setting _shutdown before flushing might prevent some exporters from flushing
+    // Waiting until flushing is complete might allow another flush to occur during shutdown
+    const flushPromise = this.forceFlush();
+    this._shutdown = true;
+    await flushPromise;
+  }
+
+  isShutdown() {
+    return this._shutdown;
+  }
+}
+
+export class ConsoleMetricExporter extends MetricExporter {
+  async export(_batch: MetricData[]) {
+    throw new Error('Method not implemented');
+  }
+
+  getPreferredAggregationTemporality() {
+    return AggregationTemporality.CUMULATIVE;
+  }
+
+  // nothing to do
+  async forceFlush() {}
+}

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

@@ -0,0 +1,24 @@
+/*
+ * 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 { MetricData } from './MetricData';
+
+/**
+ * This is a public interface that represent an export state of a MetricReader.
+ */
+export interface MetricProducer {
+  collect(): Promise<MetricData[]>;
+}