From 36f4d0a9ab8d304c0797c3afa99288932b9de1ad Mon Sep 17 00:00:00 2001 From: nikhilbhatia08 Date: Thu, 24 Sep 2026 01:42:31 +0530 Subject: [PATCH 1/3] Fix spatial re-aggregation for asynchronous instruments --- CHANGELOG.md | 9 + .../sdk/metrics/state/async_metric_storage.h | 71 ++++++- sdk/src/metrics/meter.cc | 2 +- sdk/test/metrics/async_metric_storage_test.cc | 177 +++++++++++++++++- sdk/test/metrics/sum_aggregation_test.cc | 74 ++++++++ 5 files changed, 321 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e203290157..4aab77c08c 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h index b1c31cf324..a0de066ce3 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h @@ -6,9 +6,11 @@ #include #include #include +#include #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 attributes_processor, #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW ExemplarFilterType exemplar_filter_type, nostd::shared_ptr &&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(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,39 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora exemplar_reservoir_->OfferMeasurement(measurement.second, measurement.first, {}); } #endif + MetricAttributes attributes = FilterAttributes(measurement.first); + observations + .GetOrSetDefault(std::move(attributes), + [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 +165,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 attributes_processor_; std::unique_ptr cumulative_hash_map_; std::unique_ptr delta_hash_map_; std::mutex hashmap_lock_; diff --git a/sdk/src/metrics/meter.cc b/sdk/src/metrics/meter.cc index 0abd75265e..12280976dd 100644 --- a/sdk/src/metrics/meter.cc +++ b/sdk/src/metrics/meter.cc @@ -615,7 +615,7 @@ std::unique_ptr Meter::RegisterAsyncMetricStorage( { WarnOnDuplicateInstrument(GetInstrumentationScope(), storage_registry_, view_instr_desc); async_storage = std::shared_ptr(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(), diff --git a/sdk/test/metrics/async_metric_storage_test.cc b/sdk/test/metrics/async_metric_storage_test.cc index 54eec782ec..49dde2ad94 100644 --- a/sdk/test/metrics/async_metric_storage_test.cc +++ b/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(), #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(), #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(), #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 +{}; + +// 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 collector(new MockCollectorHandle(temporality)); + std::vector> 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 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 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(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( + 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(data.value_), get_count_v1 + get_count_v2); + } + else + { + EXPECT_EQ(opentelemetry::nostd::get(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 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(data_attr.point_data); + EXPECT_EQ(data_attr.attributes.end(), data_attr.attributes.find("version")); + if (opentelemetry::nostd::get( + data_attr.attributes.find("RequestType")->second) == "GET") + { + if (temporality == AggregationTemporality::kCumulative) + { + EXPECT_EQ(opentelemetry::nostd::get(data.value_), + get_count_v1_2 + get_count_v2_2); + } + else + { + EXPECT_EQ(opentelemetry::nostd::get(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(data.value_), put_count_v1_2); + } + else + { + EXPECT_EQ(opentelemetry::nostd::get(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 collector( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors; + collectors.push_back(collector); + + // Empty allow list - every attribute is dropped. + std::shared_ptr 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 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(data_attr.point_data); + ++collected_points; + EXPECT_EQ(0, data_attr.attributes.size()); + EXPECT_DOUBLE_EQ(opentelemetry::nostd::get(data.value_), -1.0); + } + return true; + }); + EXPECT_EQ(collected_points, 1); +} + } // namespace diff --git a/sdk/test/metrics/sum_aggregation_test.cc b/sdk/test/metrics/sum_aggregation_test.cc index 03119906d1..9bd25e8c70 100644 --- a/sdk/test/metrics/sum_aggregation_test.cc +++ b/sdk/test/metrics/sum_aggregation_test.cc @@ -13,7 +13,9 @@ #include "opentelemetry/common/macros.h" #include "opentelemetry/context/context.h" +#include "opentelemetry/metrics/async_instruments.h" #include "opentelemetry/metrics/meter.h" +#include "opentelemetry/metrics/observer_result.h" #include "opentelemetry/metrics/sync_instruments.h" #include "opentelemetry/nostd/function_ref.h" #include "opentelemetry/nostd/shared_ptr.h" @@ -411,6 +413,78 @@ TEST(CounterToSumFilterAttributesWithCardinalityLimit, Double) } } +namespace +{ +// Observes two different "version" dimensions for the same "attr1" value. +void ObservableCounterCallback(opentelemetry::metrics::ObserverResult observer, void * /* state */) +{ + auto observer_double = opentelemetry::nostd::get< + opentelemetry::nostd::shared_ptr>>(observer); + observer_double->Observe(1.0, + std::map{{"attr1", "val1"}, {"version", "1"}}); + observer_double->Observe(2.0, + std::map{{"attr1", "val1"}, {"version", "2"}}); +} +} // namespace + +// A view which drops the "version" dimension collapses both observations onto the same +// attribute set. Being an additive instrument, they must be summed up (1 + 2), and not +// reported as the last observed value. +TEST(AsyncCounterToSumFilterAttributes, Double) +{ + MeterProvider mp; + auto m = mp.GetMeter("meter1", "version1", "schema1"); + std::string instrument_unit = "ms"; + std::string instrument_name = "observable_counter1"; + std::string instrument_desc = "observable counter metrics"; + + opentelemetry::sdk::metrics::FilterAttributeMap allowedattr; + allowedattr["attr1"] = true; + std::unique_ptr attrproc{ + new opentelemetry::sdk::metrics::FilteringAttributesProcessor(allowedattr)}; + + std::shared_ptr dummy_aggregation_config{ + new opentelemetry::sdk::metrics::AggregationConfig}; + std::unique_ptr exporter(new MockMetricExporter()); + std::shared_ptr reader{new MockMetricReader(std::move(exporter))}; + mp.AddMetricReader(reader); + + std::unique_ptr view{new View("view1", "view1_description", AggregationType::kSum, + dummy_aggregation_config, std::move(attrproc))}; + std::unique_ptr instrument_selector{ + new InstrumentSelector(InstrumentType::kObservableCounter, instrument_name, instrument_unit)}; + std::unique_ptr meter_selector{new MeterSelector("meter1", "version1", "schema1")}; + mp.AddView(std::move(instrument_selector), std::move(meter_selector), std::move(view)); + + auto c = m->CreateDoubleObservableCounter(instrument_name, instrument_desc, instrument_unit); + c->AddCallback(ObservableCounterCallback, nullptr); + + size_t collected_points = 0; + reader->Collect([&](ResourceMetrics &rm) { + for (const ScopeMetrics &smd : rm.scope_metric_data_) + { + for (const MetricData &md : smd.metric_data_) + { + EXPECT_EQ(1, md.point_data_attr_.size()); + for (const PointDataAttributes &dp : md.point_data_attr_) + { + ++collected_points; + EXPECT_EQ(3.0, opentelemetry::nostd::get( + opentelemetry::nostd::get(dp.point_data).value_)); + // Only the attribute allowed by the view is reported. + EXPECT_EQ(1, dp.attributes.size()); + EXPECT_NE(dp.attributes.end(), dp.attributes.find("attr1")); + EXPECT_EQ(dp.attributes.end(), dp.attributes.find("version")); + } + } + } + return true; + }); + EXPECT_EQ(1, collected_points); + + c->RemoveCallback(ObservableCounterCallback, nullptr); +} + namespace { From c9f670367cde1194a177cee6bfc831dd76f2eed9 Mon Sep 17 00:00:00 2001 From: nikhilbhatia08 Date: Thu, 24 Sep 2026 02:16:03 +0530 Subject: [PATCH 2/3] cppcheck warning fix --- .../opentelemetry/sdk/metrics/state/async_metric_storage.h | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h index a0de066ce3..31840d555e 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h @@ -86,9 +86,8 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora exemplar_reservoir_->OfferMeasurement(measurement.second, measurement.first, {}); } #endif - MetricAttributes attributes = FilterAttributes(measurement.first); observations - .GetOrSetDefault(std::move(attributes), + .GetOrSetDefault(FilterAttributes(measurement.first), [this]() { return DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_); From 7f6100f9f221b825fc48b7015f97cfa04ff45266 Mon Sep 17 00:00:00 2001 From: nikhilbhatia08 Date: Thu, 24 Sep 2026 02:23:26 +0530 Subject: [PATCH 3/3] Re-trigger CI