Sync and Async Instruments SDK (#1184)

api/include/opentelemetry/metrics/async_instruments.h

@@ -4,7 +4,7 @@
 #pragma once
 #ifndef ENABLE_METRICS_PREVIEW
 
-#  include "observer_result.h"
+#  include "opentelemetry/metrics/observer_result.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace metrics

api/include/opentelemetry/metrics/sync_instruments.h

@@ -8,6 +8,7 @@
 #  include "opentelemetry/common/key_value_iterable_view.h"
 #  include "opentelemetry/nostd/span.h"
 #  include "opentelemetry/nostd/string_view.h"
+#  include "opentelemetry/nostd/type_traits.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace metrics

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

@@ -0,0 +1,179 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#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/metrics/instruments.h"
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+template <class T>
+class Asynchronous
+{
+public:
+  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)
+  {}
+
+protected:
+  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_;
+};
+
+class LongObservableCounter : public opentelemetry::metrics::ObservableCounter<long>,
+                              public Asynchronous<long>
+{
+public:
+  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)
+
+  {}
+};
+
+class DoubleObservableCounter : public opentelemetry::metrics::ObservableCounter<double>,
+                                public Asynchronous<double>
+{
+public:
+  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)
+
+  {}
+};
+
+class LongObservableGauge : public opentelemetry::metrics::ObservableGauge<long>,
+                            public Asynchronous<long>
+{
+public:
+  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)
+
+  {}
+};
+
+class DoubleObservableGauge : public opentelemetry::metrics::ObservableGauge<double>,
+                              public Asynchronous<double>
+{
+public:
+  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)
+
+  {}
+};
+
+class LongObservableUpDownCounter : public opentelemetry::metrics::ObservableUpDownCounter<long>,
+                                    public Asynchronous<long>
+{
+public:
+  LongObservableUpDownCounter(
+      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)
+
+  {}
+};
+
+class DoubleObservableUpDownCounter
+    : public opentelemetry::metrics::ObservableUpDownCounter<double>,
+      public Asynchronous<double>
+{
+public:
+  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)
+  {}
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

sdk/include/opentelemetry/sdk/metrics/measurement_processor.h

@@ -0,0 +1,108 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+
+#  include "opentelemetry/common/key_value_iterable_view.h"
+#  include "opentelemetry/sdk/metrics/instruments.h"
+#  include "opentelemetry/sdk/metrics/metric_reader.h"
+#  include "opentelemetry/sdk/metrics/state/sync_metric_storage.h"
+
+#  include <map>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+static std::size_t MakeKey(const MetricReader &metric_reader)
+{
+  return reinterpret_cast<std::size_t>(&metric_reader);
+}
+class MeasurementProcessor
+{
+public:
+  virtual void RecordLong(long value) noexcept = 0;
+
+  virtual void RecordLong(long value,
+                          const opentelemetry::common::KeyValueIterable &attributes) noexcept = 0;
+
+  virtual void RecordDouble(double value) noexcept = 0;
+
+  virtual void RecordDouble(double value,
+                            const opentelemetry::common::KeyValueIterable &attributes) noexcept = 0;
+
+  virtual bool Collect(MetricReader &reader,
+                       AggregationTemporarily aggregation_temporarily,
+                       nostd::function_ref<bool(MetricData)> callback) noexcept = 0;
+};
+
+class DefaultMeasurementProcessor : public MeasurementProcessor
+{
+
+public:
+  bool AddMetricStorage(const MetricReader &reader)
+  {
+    // TBD = check if already present.
+    metric_storages_[MakeKey(reader)] = std::unique_ptr<SyncMetricStorage>(new SyncMetricStorage());
+    return true;
+  }
+
+  virtual void RecordLong(long value) noexcept override
+  {
+    for (const auto &kv : metric_storages_)
+    {
+      kv.second->RecordLong(value);
+    }
+  }
+
+  virtual void RecordLong(
+      long value,
+      const opentelemetry::common::KeyValueIterable &attributes) noexcept override
+  {
+    for (const auto &kv : metric_storages_)
+    {
+      kv.second->RecordLong(value, attributes);
+    }
+  }
+
+  virtual void RecordDouble(double value) noexcept override
+  {
+    for (const auto &kv : metric_storages_)
+    {
+      kv.second->RecordDouble(value);
+    }
+  }
+
+  virtual void RecordDouble(
+      double value,
+      const opentelemetry::common::KeyValueIterable &attributes) noexcept override
+  {
+    for (const auto &kv : metric_storages_)
+    {
+      kv.second->RecordDouble(value, attributes);
+    }
+  }
+
+  bool Collect(MetricReader &reader,
+               AggregationTemporarily aggregation_temporarily,
+               nostd::function_ref<bool(MetricData)> callback) noexcept override
+  {
+    auto i = metric_storages_.find(MakeKey(reader));
+    if (i != metric_storages_.end())
+    {
+      return i->second->Collect(aggregation_temporarily, callback);
+    }
+    return false;
+  }
+
+private:
+  std::map<std::size_t, std::unique_ptr<SyncMetricStorage>> metric_storages_;
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif

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

@@ -6,6 +6,7 @@
 #  include <chrono>
 #  include "opentelemetry/metrics/meter.h"
 #  include "opentelemetry/sdk/instrumentationlibrary/instrumentation_library.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"
@@ -94,13 +95,16 @@ class Meter final : public opentelemetry::metrics::Meter
       nostd::string_view unit        = "") noexcept override;
 
   /** Returns the associated instruementation library */
-  const sdk::instrumentationlibrary::InstrumentationLibrary &GetInstrumentationLibrary()
+  const sdk::instrumentationlibrary::InstrumentationLibrary *GetInstrumentationLibrary()
       const noexcept;
 
+  /** Returns the associated measurement processor */
+  MeasurementProcessor *GetMeasurementProcessor() const noexcept;
+
 private:
   // order of declaration is important here - instrumentation library should destroy after
   // meter-context.
-  std::shared_ptr<sdk::instrumentationlibrary::InstrumentationLibrary> instrumentation_library_;
+  std::unique_ptr<sdk::instrumentationlibrary::InstrumentationLibrary> instrumentation_library_;
   std::shared_ptr<sdk::metrics::MeterContext> context_;
 };
 }  // namespace metrics

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

@@ -46,6 +46,12 @@ class MeterContext
    */
   const opentelemetry::sdk::resource::Resource &GetResource() const noexcept;
 
+  /**
+   * Obtain the reference of measurement_processor.
+   *
+   */
+  MeasurementProcessor *GetMeasurementProcessor() const noexcept;
+
   /**
    * Attaches a metric exporter to list of configured exporters for this Meter context.
    * @param exporter The metric exporter for this meter context. This
@@ -95,6 +101,7 @@ class MeterContext
   std::vector<std::unique_ptr<MetricExporter>> exporters_;
   std::vector<std::unique_ptr<MetricReader>> readers_;
   std::unique_ptr<ViewRegistry> views_;
+  std::unique_ptr<MeasurementProcessor> measurement_processor_;
 };
 
 }  // namespace metrics

sdk/include/opentelemetry/sdk/metrics/meter_provider.h

@@ -8,6 +8,7 @@
 #  include <vector>
 #  include "opentelemetry/metrics/meter_provider.h"
 #  include "opentelemetry/nostd/shared_ptr.h"
+#  include "opentelemetry/sdk/metrics/measurement_processor.h"
 #  include "opentelemetry/sdk/metrics/meter.h"
 #  include "opentelemetry/sdk/metrics/meter_context.h"
 #  include "opentelemetry/sdk/resource/resource.h"
@@ -51,6 +52,12 @@ class MeterProvider final : public opentelemetry::metrics::MeterProvider
    */
   const sdk::resource::Resource &GetResource() const noexcept;
 
+  /**
+   * Obtain the reference of measurement processor.
+   *
+   */
+  MeasurementProcessor *GetMeasurementProcessor() const noexcept;
+
   /**
    * Attaches a metric exporter to list of configured exporters for this Meter provider.
    * @param exporter The metric exporter for this meter provider. This

sdk/include/opentelemetry/sdk/metrics/metric_exporter.h

@@ -13,6 +13,7 @@ namespace sdk
 {
 namespace metrics
 {
+
 /**
  * MetricExporter defines the interface to be used by metrics libraries to
  *  push metrics data to the OpenTelemetry exporters.
@@ -43,6 +44,9 @@ class MetricExporter
    * @return return the status of the operation.
    */
   virtual bool Shutdown() noexcept = 0;
+
+private:
+  AggregationTemporarily aggregation_temporarily;
 };
 }  // namespace metrics
 }  // namespace sdk