Fix:1674, Add Monotonic Property to Sum Aggregation, and unit tests f…

…or Up Down Counter (open-telemetry#1675)
yxue · Dec 5, 2022 · ea9d10c · ea9d10c
1 parent 2c55ec9
commit ea9d10c
Show file tree

Hide file tree

Showing 11 changed files with 429 additions and 29 deletions.
diff --git a/exporters/prometheus/include/opentelemetry/exporters/prometheus/exporter_utils.h b/exporters/prometheus/include/opentelemetry/exporters/prometheus/exporter_utils.h
@@ -49,7 +49,8 @@ class PrometheusExporterUtils
   /**
    * Translate the OTel metric type to Prometheus metric type
    */
-  static ::prometheus::MetricType TranslateType(opentelemetry::sdk::metrics::AggregationType kind);
+  static ::prometheus::MetricType TranslateType(opentelemetry::sdk::metrics::AggregationType kind,
+                                                bool is_monotonic = true);
 
   /**
    * Set metric data for:

diff --git a/exporters/prometheus/src/exporter_utils.cc b/exporters/prometheus/src/exporter_utils.cc
@@ -54,8 +54,14 @@ std::vector<prometheus_client::MetricFamily> PrometheusExporterUtils::TranslateT
         auto time          = metric_data.start_ts.time_since_epoch();
         for (const auto &point_data_attr : metric_data.point_data_attr_)
         {
-          auto kind                                = getAggregationType(point_data_attr.point_data);
-          const prometheus_client::MetricType type = TranslateType(kind);
+          auto kind         = getAggregationType(point_data_attr.point_data);
+          bool is_monotonic = true;
+          if (kind == sdk::metrics::AggregationType::kSum)
+          {
+            is_monotonic =
+                nostd::get<sdk::metrics::SumPointData>(point_data_attr.point_data).is_monotonic_;
+          }
+          const prometheus_client::MetricType type = TranslateType(kind, is_monotonic);
           metric_family.type                       = type;
           if (type == prometheus_client::MetricType::Histogram)  // Histogram
           {
@@ -85,6 +91,14 @@ std::vector<prometheus_client::MetricFamily> PrometheusExporterUtils::TranslateT
               std::vector<metric_sdk::ValueType> values{last_value_point_data.value_};
               SetData(values, point_data_attr.attributes, type, time, &metric_family);
             }
+            else if (nostd::holds_alternative<sdk::metrics::SumPointData>(
+                         point_data_attr.point_data))
+            {
+              auto sum_point_data =
+                  nostd::get<sdk::metrics::SumPointData>(point_data_attr.point_data);
+              std::vector<metric_sdk::ValueType> values{sum_point_data.value_};
+              SetData(values, point_data_attr.attributes, type, time, &metric_family);
+            }
             else
             {
               OTEL_INTERNAL_LOG_WARN(
@@ -159,16 +173,27 @@ metric_sdk::AggregationType PrometheusExporterUtils::getAggregationType(
  * Translate the OTel metric type to Prometheus metric type
  */
 prometheus_client::MetricType PrometheusExporterUtils::TranslateType(
-    metric_sdk::AggregationType kind)
+    metric_sdk::AggregationType kind,
+    bool is_monotonic)
 {
   switch (kind)
   {
     case metric_sdk::AggregationType::kSum:
-      return prometheus_client::MetricType::Counter;
+      if (!is_monotonic)
+      {
+        return prometheus_client::MetricType::Gauge;
+      }
+      else
+      {
+        return prometheus_client::MetricType::Counter;
+      }
+      break;
     case metric_sdk::AggregationType::kHistogram:
       return prometheus_client::MetricType::Histogram;
+      break;
     case metric_sdk::AggregationType::kLastValue:
       return prometheus_client::MetricType::Gauge;
+      break;
     default:
       return prometheus_client::MetricType::Untyped;
   }

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/default_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/default_aggregation.h
@@ -32,12 +32,15 @@ class DefaultAggregation
     switch (instrument_descriptor.type_)
     {
       case InstrumentType::kCounter:
-      case InstrumentType::kUpDownCounter:
       case InstrumentType::kObservableCounter:
+        return (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
+                   ? std::move(std::unique_ptr<Aggregation>(new LongSumAggregation(true)))
+                   : std::move(std::unique_ptr<Aggregation>(new DoubleSumAggregation(true)));
+      case InstrumentType::kUpDownCounter:
       case InstrumentType::kObservableUpDownCounter:
         return (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
-                   ? std::move(std::unique_ptr<Aggregation>(new LongSumAggregation()))
-                   : std::move(std::unique_ptr<Aggregation>(new DoubleSumAggregation()));
+                   ? std::move(std::unique_ptr<Aggregation>(new LongSumAggregation(false)))
+                   : std::move(std::unique_ptr<Aggregation>(new DoubleSumAggregation(false)));
         break;
       case InstrumentType::kHistogram: {
         if (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
@@ -91,16 +94,23 @@ class DefaultAggregation
           return std::unique_ptr<Aggregation>(new DoubleLastValueAggregation());
         }
         break;
-      case AggregationType::kSum:
+      case AggregationType::kSum: {
+        bool is_monotonic = true;
+        if (instrument_descriptor.type_ == InstrumentType::kUpDownCounter ||
+            instrument_descriptor.type_ == InstrumentType::kObservableUpDownCounter)
+        {
+          is_monotonic = false;
+        }
         if (instrument_descriptor.value_type_ == InstrumentValueType::kLong)
         {
-          return std::unique_ptr<Aggregation>(new LongSumAggregation());
+          return std::unique_ptr<Aggregation>(new LongSumAggregation(is_monotonic));
         }
         else
         {
-          return std::unique_ptr<Aggregation>(new DoubleSumAggregation());
+          return std::unique_ptr<Aggregation>(new DoubleSumAggregation(is_monotonic));
         }
         break;
+      }
       default:
         return DefaultAggregation::CreateAggregation(instrument_descriptor, aggregation_config);
     }

diff --git a/sdk/include/opentelemetry/sdk/metrics/aggregation/sum_aggregation.h b/sdk/include/opentelemetry/sdk/metrics/aggregation/sum_aggregation.h
@@ -17,7 +17,7 @@ namespace metrics
 class LongSumAggregation : public Aggregation
 {
 public:
-  LongSumAggregation();
+  LongSumAggregation(bool is_monotonic);
   LongSumAggregation(SumPointData &&);
   LongSumAggregation(const SumPointData &);
 
@@ -39,7 +39,7 @@ class LongSumAggregation : public Aggregation
 class DoubleSumAggregation : public Aggregation
 {
 public:
-  DoubleSumAggregation();
+  DoubleSumAggregation(bool is_monotonic);
   DoubleSumAggregation(SumPointData &&);
   DoubleSumAggregation(const SumPointData &);
 

diff --git a/sdk/include/opentelemetry/sdk/metrics/data/point_data.h b/sdk/include/opentelemetry/sdk/metrics/data/point_data.h
@@ -29,7 +29,8 @@ class SumPointData
   SumPointData &operator=(SumPointData &&) = default;
   SumPointData()                           = default;
 
-  ValueType value_ = {};
+  ValueType value_   = {};
+  bool is_monotonic_ = true;
 };
 
 class LastValuePointData

diff --git a/sdk/src/metrics/aggregation/sum_aggregation.cc b/sdk/src/metrics/aggregation/sum_aggregation.cc
@@ -3,6 +3,7 @@
 
 #ifndef ENABLE_METRICS_PREVIEW
 #  include "opentelemetry/sdk/metrics/aggregation/sum_aggregation.h"
+#  include "opentelemetry/sdk/common/global_log_handler.h"
 #  include "opentelemetry/sdk/metrics/data/point_data.h"
 #  include "opentelemetry/version.h"
 
@@ -15,9 +16,10 @@ namespace sdk
 namespace metrics
 {
 
-LongSumAggregation::LongSumAggregation()
+LongSumAggregation::LongSumAggregation(bool is_monotonic)
 {
-  point_data_.value_ = (int64_t)0;
+  point_data_.value_        = (int64_t)0;
+  point_data_.is_monotonic_ = is_monotonic;
 }
 
 LongSumAggregation::LongSumAggregation(SumPointData &&data) : point_data_{std::move(data)} {}
@@ -26,6 +28,14 @@ LongSumAggregation::LongSumAggregation(const SumPointData &data) : point_data_{d
 
 void LongSumAggregation::Aggregate(int64_t value, const PointAttributes & /* attributes */) noexcept
 {
+  if (point_data_.is_monotonic_ && value < 0)
+  {
+    OTEL_INTERNAL_LOG_WARN(
+        " LongSumAggregation::Aggregate Negative value ignored for Monotonic increasing "
+        "measurement. Value"
+        << value);
+    return;
+  }
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
   point_data_.value_ = nostd::get<int64_t>(point_data_.value_) + value;
 }
@@ -37,20 +47,19 @@ std::unique_ptr<Aggregation> LongSumAggregation::Merge(const Aggregation &delta)
           nostd::get<SumPointData>((static_cast<const LongSumAggregation &>(delta).ToPoint()))
               .value_) +
       nostd::get<int64_t>(nostd::get<SumPointData>(ToPoint()).value_);
-  std::unique_ptr<Aggregation> aggr(new LongSumAggregation());
+  std::unique_ptr<Aggregation> aggr(new LongSumAggregation(point_data_.is_monotonic_));
   static_cast<LongSumAggregation *>(aggr.get())->point_data_.value_ = merge_value;
   return aggr;
 }
 
 std::unique_ptr<Aggregation> LongSumAggregation::Diff(const Aggregation &next) const noexcept
 {
-
   int64_t diff_value =
       nostd::get<int64_t>(
           nostd::get<SumPointData>((static_cast<const LongSumAggregation &>(next).ToPoint()))
               .value_) -
       nostd::get<int64_t>(nostd::get<SumPointData>(ToPoint()).value_);
-  std::unique_ptr<Aggregation> aggr(new LongSumAggregation());
+  std::unique_ptr<Aggregation> aggr(new LongSumAggregation(point_data_.is_monotonic_));
   static_cast<LongSumAggregation *>(aggr.get())->point_data_.value_ = diff_value;
   return aggr;
 }
@@ -61,9 +70,10 @@ PointType LongSumAggregation::ToPoint() const noexcept
   return point_data_;
 }
 
-DoubleSumAggregation::DoubleSumAggregation()
+DoubleSumAggregation::DoubleSumAggregation(bool is_monotonic)
 {
-  point_data_.value_ = 0.0;
+  point_data_.value_        = 0.0;
+  point_data_.is_monotonic_ = is_monotonic;
 }
 
 DoubleSumAggregation::DoubleSumAggregation(SumPointData &&data) : point_data_(std::move(data)) {}
@@ -73,6 +83,14 @@ DoubleSumAggregation::DoubleSumAggregation(const SumPointData &data) : point_dat
 void DoubleSumAggregation::Aggregate(double value,
                                      const PointAttributes & /* attributes */) noexcept
 {
+  if (point_data_.is_monotonic_ && value < 0)
+  {
+    OTEL_INTERNAL_LOG_WARN(
+        " DoubleSumAggregation::Aggregate Negative value ignored for Monotonic increasing "
+        "measurement. Value"
+        << value);
+    return;
+  }
   const std::lock_guard<opentelemetry::common::SpinLockMutex> locked(lock_);
   point_data_.value_ = nostd::get<double>(point_data_.value_) + value;
 }
@@ -84,20 +102,19 @@ std::unique_ptr<Aggregation> DoubleSumAggregation::Merge(const Aggregation &delt
           nostd::get<SumPointData>((static_cast<const DoubleSumAggregation &>(delta).ToPoint()))
               .value_) +
       nostd::get<double>(nostd::get<SumPointData>(ToPoint()).value_);
-  std::unique_ptr<Aggregation> aggr(new DoubleSumAggregation());
+  std::unique_ptr<Aggregation> aggr(new DoubleSumAggregation(point_data_.is_monotonic_));
   static_cast<DoubleSumAggregation *>(aggr.get())->point_data_.value_ = merge_value;
   return aggr;
 }
 
 std::unique_ptr<Aggregation> DoubleSumAggregation::Diff(const Aggregation &next) const noexcept
 {
-
   double diff_value =
       nostd::get<double>(
           nostd::get<SumPointData>((static_cast<const DoubleSumAggregation &>(next).ToPoint()))
               .value_) -
       nostd::get<double>(nostd::get<SumPointData>(ToPoint()).value_);
-  std::unique_ptr<Aggregation> aggr(new DoubleSumAggregation());
+  std::unique_ptr<Aggregation> aggr(new DoubleSumAggregation(point_data_.is_monotonic_));
   static_cast<DoubleSumAggregation *>(aggr.get())->point_data_.value_ = diff_value;
   return aggr;
 }

diff --git a/sdk/test/metrics/BUILD b/sdk/test/metrics/BUILD
@@ -95,6 +95,22 @@ cc_test(
     ],
 )
 
+cc_test(
+    name = "sync_metric_storage_up_down_counter_test",
+    srcs = [
+        "sync_metric_storage_up_down_counter_test.cc",
+    ],
+    tags = [
+        "metrics",
+        "test",
+    ],
+    deps = [
+        "//sdk/src/metrics",
+        "//sdk/src/resource",
+        "@com_google_googletest//:gtest_main",
+    ],
+)
+
 cc_test(
     name = "sync_metric_storage_histogram_test",
     srcs = [

diff --git a/sdk/test/metrics/CMakeLists.txt b/sdk/test/metrics/CMakeLists.txt
@@ -8,6 +8,7 @@ foreach(
   attributes_hashmap_test
   sync_metric_storage_counter_test
   sync_metric_storage_histogram_test
+  sync_metric_storage_up_down_counter_test
   async_metric_storage_test
   multi_metric_storage_test
   observer_result_test

diff --git a/sdk/test/metrics/aggregation_test.cc b/sdk/test/metrics/aggregation_test.cc
@@ -14,7 +14,7 @@ using namespace opentelemetry::sdk::metrics;
 namespace nostd = opentelemetry::nostd;
 TEST(Aggregation, LongSumAggregation)
 {
-  LongSumAggregation aggr;
+  LongSumAggregation aggr(true);
   auto data = aggr.ToPoint();
   ASSERT_TRUE(nostd::holds_alternative<SumPointData>(data));
   auto sum_data = nostd::get<SumPointData>(data);
@@ -28,7 +28,7 @@ TEST(Aggregation, LongSumAggregation)
 
 TEST(Aggregation, DoubleSumAggregation)
 {
-  DoubleSumAggregation aggr;
+  DoubleSumAggregation aggr(true);
   auto data = aggr.ToPoint();
   ASSERT_TRUE(nostd::holds_alternative<SumPointData>(data));
   auto sum_data = nostd::get<SumPointData>(data);

diff --git a/sdk/test/metrics/sync_metric_storage_counter_test.cc b/sdk/test/metrics/sync_metric_storage_counter_test.cc
@@ -38,7 +38,7 @@ class MockCollectorHandle : public CollectorHandle
 class WritableMetricStorageTestFixture : public ::testing::TestWithParam<AggregationTemporality>
 {};
 
-TEST_P(WritableMetricStorageTestFixture, LongSumAggregation)
+TEST_P(WritableMetricStorageTestFixture, LongCounterSumAggregation)
 {
   AggregationTemporality temporality  = GetParam();
   auto sdk_start_ts                   = std::chrono::system_clock::now();
@@ -106,7 +106,6 @@ TEST_P(WritableMetricStorageTestFixture, LongSumAggregation)
     expected_total_get_requests = 0;
     expected_total_put_requests = 0;
   }
-
   // collect one more time.
   collection_ts    = std::chrono::system_clock::now();
   count_attributes = 0;
@@ -172,7 +171,7 @@ INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestLong,
                          ::testing::Values(AggregationTemporality::kCumulative,
                                            AggregationTemporality::kDelta));
 
-TEST_P(WritableMetricStorageTestFixture, DoubleSumAggregation)
+TEST_P(WritableMetricStorageTestFixture, DoubleCounterSumAggregation)
 {
   AggregationTemporality temporality = GetParam();
   auto sdk_start_ts                  = std::chrono::system_clock::now();