[Metrics SDK] Fix spatial re-aggregation for asynchronous instruments

CHANGELOG.md

@@ -15,6 +15,15 @@ Increment the:
 
 ## [Unreleased]
 
+* [SDK] Fix spatial re-aggregation for asynchronous instruments. A view which
+  drops attributes is now applied to asynchronous instruments as well, and the
+  observations which collapse onto the same attribute set are re-aggregated
+  (summed up for the additive instruments) instead of the last observation
+  overwriting the previous ones.
+  Note that `AsyncMetricStorage`'s constructor now takes the view's
+  `AttributesProcessor`, matching `SyncMetricStorage`.
+  [#1724](https://github.com/open-telemetry/opentelemetry-cpp/issues/1724)
+
 * [EXAMPLES] Fix random attribute selection in metrics foo example to include
   all key-value pairs
   [#4585](https://github.com/open-telemetry/opentelemetry-cpp/pull/4585)

sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h

@@ -6,9 +6,11 @@
 #include <memory>
 #include <mutex>
 #include <unordered_map>
+#include <utility>
 
 #include "opentelemetry/nostd/shared_ptr.h"
 #include "opentelemetry/sdk/common/attributemap_hash.h"
+#include "opentelemetry/sdk/metrics/aggregation/aggregation.h"
 #include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
 #include "opentelemetry/sdk/metrics/aggregation/default_aggregation.h"
 
@@ -37,6 +39,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
 public:
   AsyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
                      const AggregationType aggregation_type,
+                     std::shared_ptr<const AttributesProcessor> attributes_processor,
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
                      ExemplarFilterType exemplar_filter_type,
                      nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
@@ -45,6 +48,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
       : instrument_descriptor_(instrument_descriptor),
         aggregation_type_{aggregation_type},
         aggregation_config_{AggregationConfig::GetOrDefault(aggregation_config)},
+        attributes_processor_{std::move(attributes_processor)},
         cumulative_hash_map_(
             std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
         delta_hash_map_(
@@ -68,6 +72,12 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     const bool offer_exemplars =
         ExemplarFilterEnabled(exemplar_filter_type_, opentelemetry::context::Context{});
 #endif
+
+    // The view may drop attributes (spatial dimensions), so several of the observed
+    // measurements can collapse onto the same attribute set. Those have to be re-aggregated
+    // (summed up for the additive instruments) instead of the last observation overwriting
+    // the previous ones.
+    AttributesHashMap observations{aggregation_config_->cardinality_limit_};
     for (auto &measurement : measurements)
     {
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
@@ -76,27 +86,38 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
         exemplar_reservoir_->OfferMeasurement(measurement.second, measurement.first, {});
       }
 #endif
+      observations
+          .GetOrSetDefault(FilterAttributes(measurement.first),
+                           [this]() {
+                             return DefaultAggregation::CreateAggregation(aggregation_type_,
+                                                                          instrument_descriptor_);
+                           })
+          ->Aggregate(measurement.second);
+    }
 
-      auto aggr = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_);
-      aggr->Aggregate(measurement.second);
-      auto prev = cumulative_hash_map_->Get(measurement.first);
+    observations.GetAllEntries([this](const MetricAttributes &attributes,
+                                      Aggregation &aggregation) {
+      auto aggr = DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_,
+                                                       aggregation);
+      auto prev = cumulative_hash_map_->Get(attributes);
       if (prev)
       {
         auto delta = prev->Diff(*aggr);
         // 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));
+        cumulative_hash_map_->Set(attributes, std::move(aggr));
+        delta_hash_map_->Set(attributes, std::move(delta));
       }
       else
       {
         // store received value in cumulative and delta map.
         cumulative_hash_map_->Set(
-            measurement.first,
+            attributes,
             DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, *aggr));
-        delta_hash_map_->Set(measurement.first, std::move(aggr));
+        delta_hash_map_->Set(attributes, std::move(aggr));
       }
-    }
+      return true;
+    });
   }
 
   void RecordLong(
@@ -143,9 +164,42 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
   }
 
 private:
+  /**
+   * Returns a copy of the observed attributes with the attributes dropped by the view removed.
+   */
+  MetricAttributes FilterAttributes(const MetricAttributes &attributes) const noexcept
+  {
+    MetricAttributes filtered(attributes);
+    if (!attributes_processor_)
+    {
+      return filtered;
+    }
+
+    bool dropped = false;
+    for (auto iter = filtered.begin(); iter != filtered.end();)
+    {
+      if (attributes_processor_->isPresent(iter->first))
+      {
+        ++iter;
+      }
+      else
+      {
+        iter    = filtered.erase(iter);
+        dropped = true;
+      }
+    }
+
+    if (dropped)
+    {
+      filtered.UpdateHash();
+    }
+    return filtered;
+  }
+
   InstrumentDescriptor instrument_descriptor_;
   AggregationType aggregation_type_;
   const AggregationConfig *aggregation_config_;
+  std::shared_ptr<const AttributesProcessor> attributes_processor_;
   std::unique_ptr<AttributesHashMap> cumulative_hash_map_;
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
   std::mutex hashmap_lock_;

sdk/src/metrics/meter.cc

@@ -615,7 +615,7 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
         {
           WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc);
           async_storage = std::shared_ptr<AsyncMetricStorage>(new AsyncMetricStorage(
-              view_instr_desc, view.GetAggregationType(),
+              view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
               exemplar_filter_type,
               GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),

sdk/test/metrics/async_metric_storage_test.cc

@@ -68,7 +68,7 @@ TEST_P(AsyncWritableMetricStorageTestFixture, TestAggregation)
   collectors.push_back(collector);
 
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kSum,
+      instr_desc, AggregationType::kSum, std::make_shared<DefaultAttributesProcessor>(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -163,7 +163,7 @@ TEST_P(WritableMetricStorageTestUpDownFixture, TestAggregation)
   collectors.push_back(collector);
 
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kDefault,
+      instr_desc, AggregationType::kDefault, std::make_shared<DefaultAttributesProcessor>(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -259,7 +259,7 @@ TEST_P(WritableMetricStorageTestObservableGaugeFixture, TestAggregation)
   collectors.push_back(collector);
 
   opentelemetry::sdk::metrics::AsyncMetricStorage storage(
-      instr_desc, AggregationType::kLastValue,
+      instr_desc, AggregationType::kLastValue, std::make_shared<DefaultAttributesProcessor>(),
 #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
       ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
 #endif
@@ -322,4 +322,175 @@ INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestObservableGaugeFixtureLong,
                          ::testing::Values(AggregationTemporality::kCumulative,
                                            AggregationTemporality::kDelta));
 
+class WritableMetricStorageTestFilteredAttributesFixture
+    : public ::testing::TestWithParam<AggregationTemporality>
+{};
+
+// Several observations can collapse to the same attribute set once the view drops
+// some of the spatial dimensions. For additive instruments they have to be summed up,
+// instead of the last observation overwriting the previous ones.
+TEST_P(WritableMetricStorageTestFilteredAttributesFixture, TestAggregation)
+{
+  AggregationTemporality temporality = GetParam();
+
+  InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter,
+                                     InstrumentValueType::kLong};
+
+  auto sdk_start_ts  = std::chrono::system_clock::now();
+  auto collection_ts = std::chrono::system_clock::now() + std::chrono::seconds(5);
+
+  std::shared_ptr<CollectorHandle> collector(new MockCollectorHandle(temporality));
+  std::vector<std::shared_ptr<CollectorHandle>> collectors;
+  collectors.push_back(collector);
+
+  // The view only keeps "RequestType", so the "version" dimension is dropped.
+  FilterAttributeMap allowed_attributes;
+  allowed_attributes["RequestType"] = true;
+  std::shared_ptr<const AttributesProcessor> attributes_processor{
+      new FilteringAttributesProcessor(allowed_attributes)};
+
+  opentelemetry::sdk::metrics::AsyncMetricStorage storage(
+      instr_desc, AggregationType::kSum, attributes_processor,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+      ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
+#endif
+      nullptr);
+
+  int64_t get_count_v1 = 20;
+  int64_t get_count_v2 = 10;
+  int64_t put_count_v1 = 5;
+
+  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements1 = {
+      {{{"RequestType", "GET"}, {"version", "1"}}, get_count_v1},
+      {{{"RequestType", "GET"}, {"version", "2"}}, get_count_v2},
+      {{{"RequestType", "PUT"}, {"version", "1"}}, put_count_v1}};
+  storage.RecordLong(measurements1,
+                     opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
+
+  size_t collected_points = 0;
+  storage.Collect(
+      collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData &metric_data) {
+        for (const auto &data_attr : metric_data.point_data_attr_)
+        {
+          const auto &data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
+          // "version" must have been dropped by the view.
+          EXPECT_EQ(data_attr.attributes.end(), data_attr.attributes.find("version"));
+          ++collected_points;
+          if (opentelemetry::nostd::get<std::string>(
+                  data_attr.attributes.find("RequestType")->second) == "GET")
+          {
+            // Both GET observations collapse onto the same attribute set, and are summed up.
+            EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_), get_count_v1 + get_count_v2);
+          }
+          else
+          {
+            EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_), put_count_v1);
+          }
+        }
+        return true;
+      });
+  EXPECT_EQ(collected_points, 2);
+
+  // Subsequent (monotonically increasing) observations should be re-aggregated the same way,
+  // and reported as delta/cumulative as requested by the reader.
+  int64_t get_count_v1_2 = 50;
+  int64_t get_count_v2_2 = 30;
+  int64_t put_count_v1_2 = 8;
+
+  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements2 = {
+      {{{"RequestType", "GET"}, {"version", "1"}}, get_count_v1_2},
+      {{{"RequestType", "GET"}, {"version", "2"}}, get_count_v2_2},
+      {{{"RequestType", "PUT"}, {"version", "1"}}, put_count_v1_2}};
+  storage.RecordLong(measurements2,
+                     opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
+
+  storage.Collect(
+      collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData &metric_data) {
+        for (const auto &data_attr : metric_data.point_data_attr_)
+        {
+          const auto &data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
+          EXPECT_EQ(data_attr.attributes.end(), data_attr.attributes.find("version"));
+          if (opentelemetry::nostd::get<std::string>(
+                  data_attr.attributes.find("RequestType")->second) == "GET")
+          {
+            if (temporality == AggregationTemporality::kCumulative)
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_),
+                        get_count_v1_2 + get_count_v2_2);
+            }
+            else
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_),
+                        (get_count_v1_2 + get_count_v2_2) - (get_count_v1 + get_count_v2));
+            }
+          }
+          else
+          {
+            if (temporality == AggregationTemporality::kCumulative)
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_), put_count_v1_2);
+            }
+            else
+            {
+              EXPECT_EQ(opentelemetry::nostd::get<int64_t>(data.value_),
+                        put_count_v1_2 - put_count_v1);
+            }
+          }
+        }
+        return true;
+      });
+}
+
+INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestFilteredAttributesLong,
+                         WritableMetricStorageTestFilteredAttributesFixture,
+                         ::testing::Values(AggregationTemporality::kCumulative,
+                                           AggregationTemporality::kDelta));
+
+// All the dimensions of an async up-down counter can be dropped, collapsing every
+// observation onto the empty attribute set.
+TEST(WritableMetricStorageTestFilteredAttributes, TestUpDownCounterAllAttributesDropped)
+{
+  InstrumentDescriptor instr_desc = {"name", "desc", "1unit",
+                                     InstrumentType::kObservableUpDownCounter,
+                                     InstrumentValueType::kDouble};
+
+  auto sdk_start_ts  = std::chrono::system_clock::now();
+  auto collection_ts = std::chrono::system_clock::now() + std::chrono::seconds(5);
+
+  std::shared_ptr<CollectorHandle> collector(
+      new MockCollectorHandle(AggregationTemporality::kCumulative));
+  std::vector<std::shared_ptr<CollectorHandle>> collectors;
+  collectors.push_back(collector);
+
+  // Empty allow list - every attribute is dropped.
+  std::shared_ptr<const AttributesProcessor> attributes_processor{
+      new FilteringAttributesProcessor(FilterAttributeMap{})};
+
+  opentelemetry::sdk::metrics::AsyncMetricStorage storage(
+      instr_desc, AggregationType::kSum, attributes_processor,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+      ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
+#endif
+      nullptr);
+
+  std::unordered_map<MetricAttributes, double, AttributeHashGenerator> measurements = {
+      {{{"version", "1"}}, 1.0}, {{{"version", "2"}}, 2.0}, {{{"version", "3"}}, -4.0}};
+  storage.RecordDouble(measurements,
+                       opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
+
+  size_t collected_points = 0;
+  storage.Collect(
+      collector.get(), collectors, sdk_start_ts, collection_ts, [&](const MetricData &metric_data) {
+        for (const auto &data_attr : metric_data.point_data_attr_)
+        {
+          const auto &data = opentelemetry::nostd::get<SumPointData>(data_attr.point_data);
+          ++collected_points;
+          EXPECT_EQ(0, data_attr.attributes.size());
+          EXPECT_DOUBLE_EQ(opentelemetry::nostd::get<double>(data.value_), -1.0);
+        }
+        return true;
+      });
+  EXPECT_EQ(collected_points, 1);
+}
+
 }  // namespace