Fix #1588 - Observable Gauge does not reflect updated values, and send the old value always (#1641)

lalitb · web-flow · commit f21b3a009a62 · 2022-10-02T14:38:50.000-07:00
diff --git a/examples/common/metrics_foo_library/foo_library.cc b/examples/common/metrics_foo_library/foo_library.cc
@@ -39,13 +39,16 @@ class MeasurementFetcher
     if (nostd::holds_alternative<
             nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>>>(observer_result))
     {
-      double val = (rand() % 700) + 1.1;
+      double random_incr = (rand() % 5) + 1.1;
+      value_ += random_incr;
       nostd::get<nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>>>(
           observer_result)
-          ->Observe(val /*, labelkv */);
+          ->Observe(value_ /*, labelkv */);
     }
   }
+  static double value_;
 };
+double MeasurementFetcher::value_ = 0.0;
 }  // namespace
 
 void foo_library::counter_example(const std::string &name)
diff --git a/examples/metrics_simple/metrics_ostream.cc b/examples/metrics_simple/metrics_ostream.cc
@@ -62,7 +62,7 @@ void initMetrics(const std::string &name)
   std::unique_ptr<metric_sdk::MeterSelector> observable_meter_selector{
       new metric_sdk::MeterSelector(name, version, schema)};
   std::unique_ptr<metric_sdk::View> observable_sum_view{
-      new metric_sdk::View{name, "description", metric_sdk::AggregationType::kSum}};
+      new metric_sdk::View{name, "test_description", metric_sdk::AggregationType::kSum}};
   p->AddView(std::move(observable_instrument_selector), std::move(observable_meter_selector),
              std::move(observable_sum_view));
 
diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h
@@ -43,7 +43,9 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
   void Record(const std::unordered_map<MetricAttributes, T, AttributeHashGenerator> &measurements,
               opentelemetry::common::SystemTimestamp /* observation_time */) noexcept
   {
-    // process the read measurements - aggregate and store in hashmap
+    // Async counter always record monotonically increasing values, and the
+    // exporter/reader can request either for delta or cumulative value.
+    // So we convert the async counter value to delta before passing it to temporal storage.
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(hashmap_lock_);
     for (auto &measurement : measurements)
     {
@@ -53,13 +55,14 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
       if (prev)
       {
         auto delta = prev->Diff(*aggr);
-        cumulative_hash_map_->Set(measurement.first,
-                                  DefaultAggregation::CloneAggregation(
-                                      aggregation_type_, instrument_descriptor_, *delta));
+        // store received value in cumulative map, and the diff in delta map (to pass it to temporal
+        // storage)
+        cumulative_hash_map_->Set(measurement.first, std::move(aggr));
         delta_hash_map_->Set(measurement.first, std::move(delta));
       }
       else
       {
+        // store received value in cumulative and delta map.
         cumulative_hash_map_->Set(
             measurement.first,
             DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr));
diff --git a/sdk/src/metrics/meter.cc b/sdk/src/metrics/meter.cc
@@ -293,7 +293,6 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
 std::vector<MetricData> Meter::Collect(CollectorHandle *collector,
                                        opentelemetry::common::SystemTimestamp collect_ts) noexcept
 {
-
   observable_registry_->Observe(collect_ts);
   std::vector<MetricData> metric_data_list;
   auto ctx = meter_context_.lock();
diff --git a/sdk/src/metrics/state/temporal_metric_storage.cc b/sdk/src/metrics/state/temporal_metric_storage.cc
@@ -34,6 +34,7 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,
   opentelemetry::common::SystemTimestamp last_collection_ts = sdk_start_ts;
   AggregationTemporality aggregation_temporarily =
       collector->GetAggregationTemporality(instrument_descriptor_.type_);
+
   if (delta_metrics->Size())
   {
     for (auto &col : collectors)
diff --git a/sdk/test/metrics/async_metric_storage_test.cc b/sdk/test/metrics/async_metric_storage_test.cc
@@ -107,12 +107,12 @@ TEST_P(WritableMetricStorageTestFixture, TestAggregation)
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
       instr_desc, AggregationType::kSum, default_attributes_processor.get(),
       std::shared_ptr<opentelemetry::sdk::metrics::AggregationConfig>{});
-  long get_count                                                                  = 20l;
-  long put_count                                                                  = 10l;
-  size_t attribute_count                                                          = 2;
-  std::unordered_map<MetricAttributes, long, AttributeHashGenerator> measurements = {
-      {{{"RequestType", "GET"}}, get_count}, {{{"RequestType", "PUT"}}, put_count}};
-  storage.RecordLong(measurements,
+  long get_count1                                                                  = 20l;
+  long put_count1                                                                  = 10l;
+  size_t attribute_count                                                           = 2;
+  std::unordered_map<MetricAttributes, long, AttributeHashGenerator> measurements1 = {
+      {{{"RequestType", "GET"}}, get_count1}, {{{"RequestType", "PUT"}}, put_count1}};
+  storage.RecordLong(measurements1,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
 
   storage.Collect(collector.get(), collectors, sdk_start_ts, collection_ts,
@@ -123,20 +123,57 @@ TEST_P(WritableMetricStorageTestFixture, TestAggregation)
                       if (opentelemetry::nostd::get<std::string>(
                               data_attr.attributes.find("RequestType")->second) == "GET")
                       {
-                        EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), get_count);
+                        EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), get_count1);
                       }
                       else if (opentelemetry::nostd::get<std::string>(
                                    data_attr.attributes.find("RequestType")->second) == "PUT")
                       {
-                        EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), put_count);
+                        EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), put_count1);
                       }
                     }
                     return true;
                   });
   // subsequent recording after collection shouldn't fail
-  storage.RecordLong(measurements,
+  // monotonic increasing values;
+  long get_count2 = 50l;
+  long put_count2 = 70l;
+
+  std::unordered_map<MetricAttributes, long, AttributeHashGenerator> measurements2 = {
+      {{{"RequestType", "GET"}}, get_count2}, {{{"RequestType", "PUT"}}, put_count2}};
+  storage.RecordLong(measurements2,
                      opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
-  EXPECT_EQ(MeasurementFetcher::number_of_attributes, attribute_count);
+  storage.Collect(
+      collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData data) {
+        for (auto data_attr : data.point_data_attr_)
+        {
+          auto data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
+          if (opentelemetry::nostd::get<std::string>(
+                  data_attr.attributes.find("RequestType")->second) == "GET")
+          {
+            if (temporality == AggregationTemporality::kCumulative)
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), get_count2);
+            }
+            else
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), get_count2 - get_count1);
+            }
+          }
+          else if (opentelemetry::nostd::get<std::string>(
+                       data_attr.attributes.find("RequestType")->second) == "PUT")
+          {
+            if (temporality == AggregationTemporality::kCumulative)
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), put_count2);
+            }
+            else
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<long>(data.value_), put_count2 - put_count1);
+            }
+          }
+        }
+        return true;
+      });
 }
 
 INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestLong,

Original file line number	Diff line number	Diff line change
`@@ -39,13 +39,16 @@ class MeasurementFetcher`
`39`	`39`	`if (nostd::holds_alternative<`
`40`	`40`	`nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>>>(observer_result))`
`41`	`41`	`{`
`42`		`- double val = (rand() % 700) + 1.1;`
	`42`	`+ double random_incr = (rand() % 5) + 1.1;`
	`43`	`+ value_ += random_incr;`
`43`	`44`	`nostd::get<nostd::shared_ptr<opentelemetry::metrics::ObserverResultT<double>>>(`
`44`	`45`	`observer_result)`
`45`		`- ->Observe(val /, labelkv /);`
	`46`	`+ ->Observe(value_ /, labelkv /);`
`46`	`47`	`}`
`47`	`48`	`}`
	`49`	`+ static double value_;`
`48`	`50`	`};`
	`51`	`+double MeasurementFetcher::value_ = 0.0;`
`49`	`52`	`} // namespace`
`50`	`53`
`51`	`54`	`void foo_library::counter_example(const std::string &name)`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,9 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora`
`43`	`43`	`void Record(const std::unordered_map<MetricAttributes, T, AttributeHashGenerator> &measurements,`
`44`	`44`	`opentelemetry::common::SystemTimestamp /* observation_time */) noexcept`
`45`	`45`	`{`
`46`		`- // process the read measurements - aggregate and store in hashmap`
	`46`	`+ // Async counter always record monotonically increasing values, and the`
	`47`	`+ // exporter/reader can request either for delta or cumulative value.`
	`48`	`+ // So we convert the async counter value to delta before passing it to temporal storage.`
`47`	`49`	`std::lock_guard<opentelemetry::common::SpinLockMutex> guard(hashmap_lock_);`
`48`	`50`	`for (auto &measurement : measurements)`
`49`	`51`	`{`
`@@ -53,13 +55,14 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora`
`53`	`55`	`if (prev)`
`54`	`56`	`{`
`55`	`57`	`auto delta = prev->Diff(*aggr);`
`56`		`- cumulative_hash_map_->Set(measurement.first,`
`57`		`- DefaultAggregation::CloneAggregation(`
`58`		`- aggregation_type_, instrument_descriptor_, *delta));`
	`58`	`+ // store received value in cumulative map, and the diff in delta map (to pass it to temporal`
	`59`	`+ // storage)`
	`60`	`+ cumulative_hash_map_->Set(measurement.first, std::move(aggr));`
`59`	`61`	`delta_hash_map_->Set(measurement.first, std::move(delta));`
`60`	`62`	`}`
`61`	`63`	`else`
`62`	`64`	`{`
	`65`	`+ // store received value in cumulative and delta map.`
`63`	`66`	`cumulative_hash_map_->Set(`
`64`	`67`	`measurement.first,`
`65`	`68`	`DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr));`
Original file line number	Diff line number	Diff line change
`@@ -293,7 +293,6 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(`
`293`	`293`	`std::vector<MetricData> Meter::Collect(CollectorHandle *collector,`
`294`	`294`	`opentelemetry::common::SystemTimestamp collect_ts) noexcept`
`295`	`295`	`{`
`296`		`-`
`297`	`296`	`observable_registry_->Observe(collect_ts);`
`298`	`297`	`std::vector<MetricData> metric_data_list;`
`299`	`298`	`auto ctx = meter_context_.lock();`
Original file line number	Diff line number	Diff line change
`@@ -34,6 +34,7 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,`
`34`	`34`	`opentelemetry::common::SystemTimestamp last_collection_ts = sdk_start_ts;`
`35`	`35`	`AggregationTemporality aggregation_temporarily =`
`36`	`36`	`collector->GetAggregationTemporality(instrument_descriptor_.type_);`
	`37`	`+`
`37`	`38`	`if (delta_metrics->Size())`
`38`	`39`	`{`
`39`	`40`	`for (auto &col : collectors)`