Enable metric collection for Async Instruments - Delta and Cumulative (…

…#1334)
open-telemetry · May 3, 2022 · 9c734b3 · 9c734b3 · github-actions · May 3, 2022
1 parent 13ac5c3
commit 9c734b3
Show file tree

Hide file tree

Showing 13 changed files with 391 additions and 48 deletions.
diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/default_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/default_aggregation.h
@@ -28,6 +28,7 @@ class DefaultAggregation
     {
       case InstrumentType::kCounter:
       case InstrumentType::kUpDownCounter:
+      case InstrumentType::kObservableCounter:
       case InstrumentType::kObservableUpDownCounter:
         return (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
                    ? std::move(std::unique_ptr<Aggregation>(new LongSumAggregation()))
@@ -90,6 +91,53 @@ class DefaultAggregation
         return DefaultAggregation::CreateAggregation(instrument_descriptor);
     }
   }
+
+  static std::unique_ptr<Aggregation> CloneAggregation(AggregationType aggregation_type,
+                                                       InstrumentDescriptor instrument_descriptor,
+                                                       const Aggregation &to_copy)
+  {
+    const PointType point_data = to_copy.ToPoint();
+    switch (aggregation_type)
+    {
+      case AggregationType::kDrop:
+        return std::unique_ptr<Aggregation>(new DropAggregation());
+      case AggregationType::kHistogram:
+        if (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
+        {
+          return std::unique_ptr<Aggregation>(
+              new LongHistogramAggregation(nostd::get<HistogramPointData>(point_data)));
+        }
+        else
+        {
+          return std::unique_ptr<Aggregation>(
+              new DoubleHistogramAggregation(nostd::get<HistogramPointData>(point_data)));
+        }
+      case AggregationType::kLastValue:
+        if (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
+        {
+          return std::unique_ptr<Aggregation>(
+              new LongLastValueAggregation(nostd::get<LastValuePointData>(point_data)));
+        }
+        else
+        {
+          return std::unique_ptr<Aggregation>(
+              new DoubleLastValueAggregation(nostd::get<LastValuePointData>(point_data)));
+        }
+      case AggregationType::kSum:
+        if (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
+        {
+          return std::unique_ptr<Aggregation>(
+              new LongSumAggregation(nostd::get<SumPointData>(point_data)));
+        }
+        else
+        {
+          return std::unique_ptr<Aggregation>(
+              new DoubleSumAggregation(nostd::get<SumPointData>(point_data)));
+        }
+      default:
+        return DefaultAggregation::CreateAggregation(instrument_descriptor);
+    }
+  }
 };
 
 }  // namespace metrics

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/drop_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/drop_aggregation.h
@@ -23,16 +23,18 @@ class DropAggregation : public Aggregation
 public:
   DropAggregation() = default;
 
+  DropAggregation(const DropPointData &) {}
+
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override {}
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override {}
 
-  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override
+  std::unique_ptr<Aggregation> Merge(const Aggregation &) const noexcept override
   {
     return std::unique_ptr<Aggregation>(new DropAggregation());
   }
 
-  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override
+  std::unique_ptr<Aggregation> Diff(const Aggregation &) const noexcept override
   {
     return std::unique_ptr<Aggregation>(new DropAggregation());
   }

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/histogram_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/histogram_aggregation.h
@@ -19,21 +19,22 @@ class LongHistogramAggregation : public Aggregation
 public:
   LongHistogramAggregation();
   LongHistogramAggregation(HistogramPointData &&);
+  LongHistogramAggregation(const HistogramPointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override;
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override {}
 
   /* Returns the result of merge of the existing aggregation with delta aggregation with same
    * boundaries */
-  virtual std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
+  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
 
   /* Returns the new delta aggregation by comparing existing aggregation with next aggregation with
    * same boundaries. Data points for `next` aggregation (sum , bucket-counts) should be more than
    * the current aggregation - which is the normal scenario as measurements values are monotonic
    * increasing.
    */
-  virtual std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
+  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
 
   PointType ToPoint() const noexcept override;
 
@@ -47,21 +48,22 @@ class DoubleHistogramAggregation : public Aggregation
 public:
   DoubleHistogramAggregation();
   DoubleHistogramAggregation(HistogramPointData &&);
+  DoubleHistogramAggregation(const HistogramPointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override {}
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override;
 
   /* Returns the result of merge of the existing aggregation with delta aggregation with same
    * boundaries */
-  virtual std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
+  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
 
   /* Returns the new delta aggregation by comparing existing aggregation with next aggregation with
    * same boundaries. Data points for `next` aggregation (sum , bucket-counts) should be more than
    * the current aggregation - which is the normal scenario as measurements values are monotonic
    * increasing.
    */
-  virtual std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
+  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
 
   PointType ToPoint() const noexcept override;
 

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/lastvalue_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/lastvalue_aggregation.h
@@ -18,14 +18,15 @@ class LongLastValueAggregation : public Aggregation
 public:
   LongLastValueAggregation();
   LongLastValueAggregation(LastValuePointData &&);
+  LongLastValueAggregation(const LastValuePointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override;
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override {}
 
-  virtual std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
+  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
 
-  virtual std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
+  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
 
   PointType ToPoint() const noexcept override;
 
@@ -39,6 +40,7 @@ class DoubleLastValueAggregation : public Aggregation
 public:
   DoubleLastValueAggregation();
   DoubleLastValueAggregation(LastValuePointData &&);
+  DoubleLastValueAggregation(const LastValuePointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override {}
 

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/sum_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/sum_aggregation.h
@@ -19,14 +19,15 @@ class LongSumAggregation : public Aggregation
 public:
   LongSumAggregation();
   LongSumAggregation(SumPointData &&);
+  LongSumAggregation(const SumPointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override;
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override {}
 
-  virtual std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
+  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
 
-  virtual std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
+  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
 
   PointType ToPoint() const noexcept override;
 
@@ -40,14 +41,15 @@ class DoubleSumAggregation : public Aggregation
 public:
   DoubleSumAggregation();
   DoubleSumAggregation(SumPointData &&);
+  DoubleSumAggregation(const SumPointData &);
 
   void Aggregate(long value, const PointAttributes &attributes = {}) noexcept override {}
 
   void Aggregate(double value, const PointAttributes &attributes = {}) noexcept override;
 
-  virtual std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
+  std::unique_ptr<Aggregation> Merge(const Aggregation &delta) const noexcept override;
 
-  virtual std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
+  std::unique_ptr<Aggregation> Diff(const Aggregation &next) const noexcept override;
 
   PointType ToPoint() const noexcept override;
 

diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h
@@ -10,6 +10,7 @@
 #  include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
 #  include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #  include "opentelemetry/sdk/metrics/state/metric_storage.h"
+#  include "opentelemetry/sdk/metrics/state/temporal_metric_storage.h"
 #  include "opentelemetry/sdk/metrics/view/attributes_processor.h"
 
 #  include <memory>
@@ -32,7 +33,8 @@ class AsyncMetricStorage : public MetricStorage
         aggregation_type_{aggregation_type},
         measurement_collection_callback_{measurement_callback},
         attributes_processor_{attributes_processor},
-        active_attributes_hashmap_(new AttributesHashMap())
+        cumulative_hash_map_(new AttributesHashMap()),
+        temporal_metric_storage_(instrument_descriptor)
   {}
 
   bool Collect(CollectorHandle *collector,
@@ -45,30 +47,42 @@ class AsyncMetricStorage : public MetricStorage
 
     // read the measurement using configured callback
     measurement_collection_callback_(ob_res);
-
+    std::shared_ptr<AttributesHashMap> delta_hash_map(new AttributesHashMap());
     // process the read measurements - aggregate and store in hashmap
     for (auto &measurement : ob_res.GetMeasurements())
     {
-      auto agg = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_);
-      agg->Aggregate(measurement.second);
-      active_attributes_hashmap_->Set(measurement.first, std::move(agg));
+      auto aggr = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_);
+      aggr->Aggregate(measurement.second);
+      auto prev = cumulative_hash_map_->Get(measurement.first);
+      if (prev)
+      {
+        auto delta = prev->Diff(*aggr);
+        cumulative_hash_map_->Set(measurement.first,
+                                  DefaultAggregation::CloneAggregation(
+                                      aggregation_type_, instrument_descriptor_, *delta));
+        delta_hash_map->Set(measurement.first, std::move(delta));
+      }
+      else
+      {
+        cumulative_hash_map_->Set(
+            measurement.first,
+            DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr));
+        delta_hash_map->Set(measurement.first, std::move(aggr));
+      }
     }
 
-    // TBD -> read aggregation from hashmap, and perform metric collection
-    MetricData metric_data;
-    if (metric_collection_callback(std::move(metric_data)))
-    {
-      return true;
-    }
-    return false;
+    return temporal_metric_storage_.buildMetrics(collector, collectors, sdk_start_ts, collection_ts,
+                                                 std::move(delta_hash_map),
+                                                 metric_collection_callback);
   }
 
 private:
   InstrumentDescriptor instrument_descriptor_;
   AggregationType aggregation_type_;
   void (*measurement_collection_callback_)(opentelemetry::metrics::ObserverResult<T> &);
   const AttributesProcessor *attributes_processor_;
-  std::unique_ptr<AttributesHashMap> active_attributes_hashmap_;
+  std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
+  TemporalMetricStorage temporal_metric_storage_;
 };
 
 }  // namespace metrics

diff --git a/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h
@@ -0,0 +1,50 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+#ifndef ENABLE_METRICS_PREVIEW
+#  include "opentelemetry/sdk/metrics/state/attributes_hashmap.h"
+#  include "opentelemetry/sdk/metrics/state/metric_collector.h"
+
+#  include <memory>
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+struct LastReportedMetrics
+{
+  std::unique_ptr<AttributesHashMap> attributes_map;
+  opentelemetry::common::SystemTimestamp collection_ts;
+};
+
+class TemporalMetricStorage
+{
+public:
+  TemporalMetricStorage(InstrumentDescriptor instrument_descriptor);
+
+  bool buildMetrics(CollectorHandle *collector,
+                    nostd::span<std::shared_ptr<CollectorHandle>> collectors,
+                    opentelemetry::common::SystemTimestamp sdk_start_ts,
+                    opentelemetry::common::SystemTimestamp collection_ts,
+                    std::shared_ptr<AttributesHashMap> delta_metrics,
+                    nostd::function_ref<bool(MetricData)> callback) noexcept;
+
+private:
+  InstrumentDescriptor instrument_descriptor_;
+
+  // unreported metrics stash for all the collectors
+  std::unordered_map<CollectorHandle *, std::list<std::shared_ptr<AttributesHashMap>>>
+      unreported_metrics_;
+  // last reported metrics stash for all the collectors.
+  std::unordered_map<CollectorHandle *, LastReportedMetrics> last_reported_metrics_;
+
+  // Lock while building metrics
+  mutable opentelemetry::common::SpinLockMutex lock_;
+};
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE
+#endif
diff --git a/sdk/src/metrics/CMakeLists.txt b/sdk/src/metrics/CMakeLists.txt
@@ -7,6 +7,7 @@ add_library(
   export/periodic_exporting_metric_reader.cc
   state/metric_collector.cc
   state/sync_metric_storage.cc
+  state/temporal_metric_storage.cc
   aggregation/histogram_aggregation.cc
   aggregation/lastvalue_aggregation.cc
   aggregation/sum_aggregation.cc

diff --git a/sdk/src/metrics/aggregation/histogram_aggregation.cc b/sdk/src/metrics/aggregation/histogram_aggregation.cc
@@ -25,6 +25,10 @@ LongHistogramAggregation::LongHistogramAggregation(HistogramPointData &&data)
     : point_data_{std::move(data)}
 {}
 
+LongHistogramAggregation::LongHistogramAggregation(const HistogramPointData &data)
+    : point_data_{data}
+{}
+
 void LongHistogramAggregation::Aggregate(long value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
@@ -83,6 +87,10 @@ DoubleHistogramAggregation::DoubleHistogramAggregation(HistogramPointData &&data
     : point_data_{std::move(data)}
 {}
 
+DoubleHistogramAggregation::DoubleHistogramAggregation(const HistogramPointData &data)
+    : point_data_{data}
+{}
+
 void DoubleHistogramAggregation::Aggregate(double value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

diff --git a/sdk/src/metrics/aggregation/lastvalue_aggregation.cc b/sdk/src/metrics/aggregation/lastvalue_aggregation.cc
@@ -19,10 +19,15 @@ LongLastValueAggregation::LongLastValueAggregation()
   point_data_.is_lastvalue_valid_ = false;
   point_data_.value_              = 0l;
 }
+
 LongLastValueAggregation::LongLastValueAggregation(LastValuePointData &&data)
     : point_data_{std::move(data)}
 {}
 
+LongLastValueAggregation::LongLastValueAggregation(const LastValuePointData &data)
+    : point_data_{data}
+{}
+
 void LongLastValueAggregation::Aggregate(long value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
@@ -71,10 +76,15 @@ DoubleLastValueAggregation::DoubleLastValueAggregation()
   point_data_.is_lastvalue_valid_ = false;
   point_data_.value_              = 0.0;
 }
+
 DoubleLastValueAggregation::DoubleLastValueAggregation(LastValuePointData &&data)
     : point_data_{std::move(data)}
 {}
 
+DoubleLastValueAggregation::DoubleLastValueAggregation(const LastValuePointData &data)
+    : point_data_{data}
+{}
+
 void DoubleLastValueAggregation::Aggregate(double value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);

diff --git a/sdk/src/metrics/aggregation/sum_aggregation.cc b/sdk/src/metrics/aggregation/sum_aggregation.cc
@@ -22,6 +22,8 @@ LongSumAggregation::LongSumAggregation()
 
 LongSumAggregation::LongSumAggregation(SumPointData &&data) : point_data_{std::move(data)} {}
 
+LongSumAggregation::LongSumAggregation(const SumPointData &data) : point_data_{data} {}
+
 void LongSumAggregation::Aggregate(long value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
@@ -64,6 +66,8 @@ DoubleSumAggregation::DoubleSumAggregation()
 
 DoubleSumAggregation::DoubleSumAggregation(SumPointData &&data) : point_data_(std::move(data)) {}
 
+DoubleSumAggregation::DoubleSumAggregation(const SumPointData &data) : point_data_(data) {}
+
 void DoubleSumAggregation::Aggregate(double value, const PointAttributes &attributes) noexcept
 {
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);