Enable metric collection from MetricReader (#1241)

exporters/ostream/include/opentelemetry/exporters/ostream/metric_exporter.h

@@ -8,6 +8,7 @@
 #  include <string>
 #  include "opentelemetry/common/spin_lock_mutex.h"
 #  include "opentelemetry/nostd/span.h"
+#  include "opentelemetry/sdk/metrics/data/metric_data.h"
 #  include "opentelemetry/sdk/metrics/instruments.h"
 #  include "opentelemetry/sdk/metrics/metric_exporter.h"
 #  include "opentelemetry/sdk/metrics/recordable.h"

sdk/include/opentelemetry/sdk/metrics/async_instruments.h

@@ -5,10 +5,8 @@
 #ifndef ENABLE_METRICS_PREVIEW
 #  include "opentelemetry/metrics/async_instruments.h"
 #  include "opentelemetry/metrics/observer_result.h"
-#  include "opentelemetry/sdk/instrumentationlibrary/instrumentation_library.h"
-#  include "opentelemetry/sdk/metrics/measurement_processor.h"
-
 #  include "opentelemetry/nostd/string_view.h"
+#  include "opentelemetry/sdk/instrumentationlibrary/instrumentation_library.h"
 #  include "opentelemetry/sdk/metrics/instruments.h"
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
@@ -23,13 +21,11 @@ class Asynchronous
   Asynchronous(nostd::string_view name,
                const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
                    *instrumentation_library,
-               MeasurementProcessor *measurement_processor,
                void (*callback)(opentelemetry::metrics::ObserverResult<T> &),
                nostd::string_view description = "",
                nostd::string_view unit        = "")
       : name_(name),
         instrumentation_library_{instrumentation_library},
-        measurement_processor_{measurement_processor},
         callback_(callback),
         description_(description),
         unit_(unit)
@@ -39,7 +35,6 @@ class Asynchronous
   std::string name_;
   const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
       *instrumentation_library_;
-  const MeasurementProcessor *measurement_processor_;
   void (*callback_)(opentelemetry::metrics::ObserverResult<T> &);
   std::string description_;
   std::string unit_;
@@ -52,16 +47,10 @@ class LongObservableCounter : public opentelemetry::metrics::ObservableCounter<l
   LongObservableCounter(nostd::string_view name,
                         const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
                             *instrumentation_library,
-                        MeasurementProcessor *measurement_processor,
                         void (*callback)(opentelemetry::metrics::ObserverResult<long> &),
                         nostd::string_view description = "",
                         nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
 
   {}
 };
@@ -73,16 +62,10 @@ class DoubleObservableCounter : public opentelemetry::metrics::ObservableCounter
   DoubleObservableCounter(nostd::string_view name,
                           const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
                               *instrumentation_library,
-                          MeasurementProcessor *measurement_processor,
                           void (*callback)(opentelemetry::metrics::ObserverResult<double> &),
                           nostd::string_view description = "",
                           nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
 
   {}
 };
@@ -94,16 +77,10 @@ class LongObservableGauge : public opentelemetry::metrics::ObservableGauge<long>
   LongObservableGauge(nostd::string_view name,
                       const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
                           *instrumentation_library,
-                      MeasurementProcessor *measurement_processor,
                       void (*callback)(opentelemetry::metrics::ObserverResult<long> &),
                       nostd::string_view description = "",
                       nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
 
   {}
 };
@@ -115,16 +92,10 @@ class DoubleObservableGauge : public opentelemetry::metrics::ObservableGauge<dou
   DoubleObservableGauge(nostd::string_view name,
                         const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
                             *instrumentation_library,
-                        MeasurementProcessor *measurement_processor,
                         void (*callback)(opentelemetry::metrics::ObserverResult<double> &),
                         nostd::string_view description = "",
                         nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
 
   {}
 };
@@ -137,16 +108,10 @@ class LongObservableUpDownCounter : public opentelemetry::metrics::ObservableUpD
       nostd::string_view name,
       const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
           *instrumentation_library,
-      MeasurementProcessor *measurement_processor,
       void (*callback)(opentelemetry::metrics::ObserverResult<long> &),
       nostd::string_view description = "",
       nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
 
   {}
 };
@@ -160,16 +125,10 @@ class DoubleObservableUpDownCounter
       nostd::string_view name,
       const opentelemetry::sdk::instrumentationlibrary::InstrumentationLibrary
           *instrumentation_library,
-      MeasurementProcessor *measurement_processor,
       void (*callback)(opentelemetry::metrics::ObserverResult<double> &),
       nostd::string_view description = "",
       nostd::string_view unit        = "")
-      : Asynchronous(name,
-                     instrumentation_library,
-                     measurement_processor,
-                     callback,
-                     description,
-                     unit)
+      : Asynchronous(name, instrumentation_library, callback, description, unit)
   {}
 };
 

sdk/include/opentelemetry/sdk/metrics/export/metric_producer.h

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/sdk/metrics/data/metric_data.h"
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+/**
+ * MetricProducer is the interface that is used to make metric data available to the
+ * OpenTelemetry exporters. Implementations should be stateful, in that each call to
+ * `Collect` will return any metric generated since the last call was made.
+ *
+ * <p>Implementations must be thread-safe.
+ */
+
+class MetricProducer
+{
+public:
+  /**
+   * The callback to be called for each metric exporter. This will only be those
+   * metrics that have been produced since the last time this method was called.
+   *
+   * @return a status of completion of method.
+   */
+  virtual bool Collect(nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

sdk/include/opentelemetry/sdk/metrics/instruments.h

@@ -54,13 +54,6 @@ struct InstrumentDescriptor
 
 using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
 
-// TBD -> Remove once MetricCollector is imoplemeted
-class MetricCollector
-{
-public:
-  AggregationTemporarily aggregation_temporarily_;
-};
-
 /*class InstrumentSelector {
 public:
 InstrumentSelector(opentelemetry::nostd::string_view name,

sdk/include/opentelemetry/sdk/metrics/meter.h

@@ -7,7 +7,6 @@
 #  include "opentelemetry/metrics/meter.h"
 #  include "opentelemetry/sdk/instrumentationlibrary/instrumentation_library.h"
 #  include "opentelemetry/sdk/metrics/instruments.h"
-#  include "opentelemetry/sdk/metrics/measurement_processor.h"
 #  include "opentelemetry/sdk/metrics/meter_context.h"
 #  include "opentelemetry/sdk/resource/resource.h"
 #  include "opentelemetry/version.h"
@@ -17,6 +16,10 @@ namespace sdk
 {
 namespace metrics
 {
+
+class MetricStorage;
+class WritableMetricStorage;
+
 class Meter final : public opentelemetry::metrics::Meter
 {
 public:
@@ -99,8 +102,10 @@ class Meter final : public opentelemetry::metrics::Meter
   const sdk::instrumentationlibrary::InstrumentationLibrary *GetInstrumentationLibrary()
       const noexcept;
 
-  /** Returns the associated measurement processor */
-  MeasurementProcessor *GetMeasurementProcessor() const noexcept;
+  /** collect metrics across all the meters **/
+  bool collect(CollectorHandle *collector,
+               opentelemetry::common::SystemTimestamp collect_ts,
+               nostd::function_ref<bool(MetricData &)> callback) noexcept;
 
 private:
   // order of declaration is important here - instrumentation library should destroy after

sdk/include/opentelemetry/sdk/metrics/meter_context.h

@@ -3,27 +3,35 @@
 
 #pragma once
 #ifndef ENABLE_METRICS_PREVIEW
-#  include <chrono>
-#  include <memory>
-#  include <vector>
-#  include "opentelemetry/sdk/metrics/metric_exporter.h"
-#  include "opentelemetry/sdk/metrics/metric_reader.h"
+
+#  include "opentelemetry/common/spin_lock_mutex.h"
+#  include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #  include "opentelemetry/sdk/metrics/view/instrument_selector.h"
 #  include "opentelemetry/sdk/metrics/view/meter_selector.h"
 #  include "opentelemetry/sdk/metrics/view/view_registry.h"
 #  include "opentelemetry/sdk/resource/resource.h"
 #  include "opentelemetry/version.h"
 
+#  include <chrono>
+#  include <memory>
+#  include <vector>
+
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
 {
 namespace metrics
 {
+
+// forward declaration
+class Meter;
+class MetricExporter;
+class MetricReader;
+
 /**
  * A class which stores the MeterProvider context.
 
  */
-class MeterContext
+class MeterContext : public std::enable_shared_from_this<MeterContext>
 {
 public:
   /**
@@ -35,7 +43,6 @@ class MeterContext
    */
   MeterContext(
       std::vector<std::unique_ptr<sdk::metrics::MetricExporter>> &&exporters,
-      std::vector<std::unique_ptr<MetricReader>> &&readers,
       std::unique_ptr<ViewRegistry> views = std::unique_ptr<ViewRegistry>(new ViewRegistry()),
       opentelemetry::sdk::resource::Resource resource =
           opentelemetry::sdk::resource::Resource::Create({})) noexcept;
@@ -47,16 +54,28 @@ class MeterContext
   const opentelemetry::sdk::resource::Resource &GetResource() const noexcept;
 
   /**
-   * Obtain the reference of measurement_processor.
+   * Obtain the View Registry configured
+   * @return The reference to view registry
+   */
+  ViewRegistry *GetViewRegistry() const noexcept;
+
+  /**
+   * Obtain the  configured meters.
    *
    */
-  MeasurementProcessor *GetMeasurementProcessor() const noexcept;
+  nostd::span<std::shared_ptr<Meter>> GetMeters() noexcept;
 
   /**
-   * Obtain the View Registry configured
-   * @return The reference to view registry
+   * Obtain the configured collectors.
+   *
    */
-  ViewRegistry *GetViewRegistry() const noexcept;
+  nostd::span<std::shared_ptr<CollectorHandle>> GetCollectors() noexcept;
+
+  /**
+   * GET SDK Start time
+   *
+   */
+  opentelemetry::common::SystemTimestamp GetSDKStartTime() noexcept;
 
   /**
    * Attaches a metric exporter to list of configured exporters for this Meter context.
@@ -91,23 +110,36 @@ class MeterContext
                std::unique_ptr<View> view) noexcept;
 
   /**
-   * Force all active Exporters and Readers to flush any buffered meter data
+   * Adds a meter to the list of configured meters.
+   *
+   * Note: This method is INTERNAL to sdk not thread safe.
+   *
+   * @param meter
+   */
+  void AddMeter(std::shared_ptr<Meter> meter);
+
+  /**
+   * Force all active Exporters and Collectors to flush any buffered meter data
    * within the given timeout.
    */
 
   bool ForceFlush(std::chrono::microseconds timeout = (std::chrono::microseconds::max)()) noexcept;
 
   /**
-   * Shutdown the Exporters and Readers associated with this meter provider.
+   * Shutdown the Exporters and Collectors associated with this meter provider.
    */
   bool Shutdown() noexcept;
 
 private:
   opentelemetry::sdk::resource::Resource resource_;
   std::vector<std::unique_ptr<MetricExporter>> exporters_;
-  std::vector<std::unique_ptr<MetricReader>> readers_;
+  std::vector<std::shared_ptr<CollectorHandle>> collectors_;
   std::unique_ptr<ViewRegistry> views_;
-  std::unique_ptr<MeasurementProcessor> measurement_processor_;
+  opentelemetry::common::SystemTimestamp sdk_start_ts_;
+  std::vector<std::shared_ptr<Meter>> meters_;
+
+  std::atomic_flag shutdown_latch_ = ATOMIC_FLAG_INIT;
+  opentelemetry::common::SpinLockMutex forceflush_lock_;
 };
 
 }  // namespace metrics