From fabd66021eebefaa67b24a9a5125020ecdd42a46 Mon Sep 17 00:00:00 2001 From: Patrick Summerer Date: Wed, 26 Aug 2026 10:17:09 +0200 Subject: [PATCH] [METRICS] Fix stale async attribute sets in cumulative exports (#4108) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Async instruments (ObservableCounter, ObservableGauge, ObservableUpDownCounter) under cumulative temporality were emitting attribute sets indefinitely after the callback stopped reporting them, violating the OTel spec requirement: "The implementation SHOULD NOT produce aggregated metric data for a previously-observed attribute set which is not observed during a successful callback." Root cause: `TemporalMetricStorage::buildMetrics()` unconditionally carried every entry from `last_reported_metrics_` into the output even when it was absent from the current delta. Fix: - Add `is_async_` flag (default false) to `TemporalMetricStorage`. The cumulative merge now skips entries not present in the current delta for async instruments, while sync instruments retain the existing carry-forward behaviour. - Pass `is_async = true` when constructing `TemporalMetricStorage` from `AsyncMetricStorage`. - Do NOT prune `cumulative_hash_map_` in `AsyncMetricStorage::Collect()` so that the absolute-value baseline is preserved across absent cycles. This ensures correct delta computation (new - last_seen, not the full new value) when an attribute set reappears after a gap — consistent with the approach taken by opentelemetry-dotnet#6883. Tests added in async_metric_storage_test.cc: - StaleAttributeSetDroppedInCumulativeExport: verifies that an attribute set absent from the callback is not emitted in subsequent cumulative exports. - AttributeReappearanceAfterGapDeltaTemporality: verifies that an attribute set reappearing after an absent cycle emits only the increment since last observed (delta = 1, not 11), confirming the baseline is correctly preserved. Fixes #4108 Co-authored-by: pranitaurlam <227409059+pranitaurlam@users.noreply.github.com> --- .../sdk/metrics/state/async_metric_storage.h | 8 +- .../metrics/state/temporal_metric_storage.h | 4 +- .../metrics/state/temporal_metric_storage.cc | 11 +- sdk/test/metrics/async_metric_storage_test.cc | 165 ++++++++++++++++++ 4 files changed, 183 insertions(+), 5 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 674863428b..dc3aa8f85f 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h @@ -53,7 +53,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora exemplar_filter_type_(exemplar_filter_type), exemplar_reservoir_(std::move(exemplar_reservoir)), #endif - temporal_metric_storage_(instrument_descriptor, aggregation_type, aggregation_config) + temporal_metric_storage_(instrument_descriptor, aggregation_type, aggregation_config, true) {} template @@ -134,6 +134,12 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora delta_metrics = std::move(delta_hash_map_); delta_hash_map_ = std::make_unique(aggregation_config_->cardinality_limit_); + // cumulative_hash_map_ is intentionally NOT pruned here. + // It preserves the last-seen absolute value for every attribute set so that + // delta computation in Record() remains correct if an attribute set reappears + // after being absent for one or more collection cycles. + // Stale entries are suppressed at export time by the is_async_ guard in + // TemporalMetricStorage::buildMetrics() instead. } auto status = diff --git a/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h b/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h index d86093c376..c0f8c77bd5 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/temporal_metric_storage.h @@ -34,7 +34,8 @@ class TemporalMetricStorage public: TemporalMetricStorage(InstrumentDescriptor instrument_descriptor, AggregationType aggregation_type, - const AggregationConfig *aggregation_config); + const AggregationConfig *aggregation_config, + bool is_async = false); bool buildMetrics(CollectorHandle *collector, nostd::span> collectors, @@ -64,6 +65,7 @@ class TemporalMetricStorage // See https://github.com/open-telemetry/opentelemetry-specification (logs/metrics // SDK specs) and issue #4062. const opentelemetry::common::SystemTimestamp instrument_creation_ts_; + bool is_async_ = false; }; } // namespace metrics } // namespace sdk diff --git a/sdk/src/metrics/state/temporal_metric_storage.cc b/sdk/src/metrics/state/temporal_metric_storage.cc index 85a0a4b3c0..60e522421f 100644 --- a/sdk/src/metrics/state/temporal_metric_storage.cc +++ b/sdk/src/metrics/state/temporal_metric_storage.cc @@ -32,11 +32,13 @@ namespace metrics TemporalMetricStorage::TemporalMetricStorage(InstrumentDescriptor instrument_descriptor, AggregationType aggregation_type, - const AggregationConfig *aggregation_config) + const AggregationConfig *aggregation_config, + bool is_async) : instrument_descriptor_(std::move(instrument_descriptor)), aggregation_type_(aggregation_type), aggregation_config_(aggregation_config), - instrument_creation_ts_(std::chrono::system_clock::now()) + instrument_creation_ts_(std::chrono::system_clock::now()), + is_async_(is_async) {} bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, @@ -159,8 +161,11 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, { merged_metrics->Set(attributes, agg->Merge(aggregation)); } - else + else if (!is_async_) { + // For sync instruments, carry forward attribute sets not observed this cycle. + // For async instruments, drop them per the spec: the SDK SHOULD NOT produce + // aggregated metric data for attribute sets not observed in the current callback. auto def_agg = DefaultAggregation::CreateAggregation( aggregation_type_, instrument_descriptor_, aggregation_config_); merged_metrics->Set(attributes, def_agg->Merge(aggregation)); diff --git a/sdk/test/metrics/async_metric_storage_test.cc b/sdk/test/metrics/async_metric_storage_test.cc index 54eec782ec..6e7ce2275a 100644 --- a/sdk/test/metrics/async_metric_storage_test.cc +++ b/sdk/test/metrics/async_metric_storage_test.cc @@ -322,4 +322,169 @@ INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestObservableGaugeFixtureLong, ::testing::Values(AggregationTemporality::kCumulative, AggregationTemporality::kDelta)); +// Regression test for https://github.com/open-telemetry/opentelemetry-cpp/issues/4108 +// +// Async instruments under cumulative temporality must NOT carry forward attribute sets that were +// not reported by the callback in the current collection cycle. +TEST(AsyncMetricStorageRegressionTest, StaleAttributeSetDroppedInCumulativeExport) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + // Some computation here + auto collection_ts = sdk_start_ts + std::chrono::seconds(5); + + std::shared_ptr collector( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors; + collectors.push_back(collector); + + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + // Collection 1: both GET and PUT reported. + std::unordered_map measurements1 = { + {{{"RequestType", "GET"}}, 10}, {{{"RequestType", "PUT"}}, 5}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int get_count = 0; + int put_count = 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 &key = opentelemetry::nostd::get( + data_attr.attributes.find("RequestType")->second); + if (key == "GET") + get_count++; + else if (key == "PUT") + put_count++; + } + return true; + }); + EXPECT_EQ(get_count, 1); + EXPECT_EQ(put_count, 1); + + // Collection 2: only GET reported – PUT is dropped by callback. + std::unordered_map measurements2 = { + {{{"RequestType", "GET"}}, 20}}; + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + get_count = 0; + put_count = 0; + int64_t get_value = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, + collection_ts + std::chrono::seconds(5), [&](const MetricData &metric_data) { + for (const auto &data_attr : metric_data.point_data_attr_) + { + const auto &key = opentelemetry::nostd::get( + data_attr.attributes.find("RequestType")->second); + if (key == "GET") + { + get_count++; + get_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + else if (key == "PUT") + { + put_count++; + } + } + return true; + }); + + // PUT must not appear – it was absent from the callback this cycle. + EXPECT_EQ(put_count, 0) << "Stale PUT attribute set must be dropped from cumulative export"; + EXPECT_EQ(get_count, 1); + EXPECT_EQ(get_value, 20); +} + +// Regression test for https://github.com/open-telemetry/opentelemetry-cpp/issues/4108 +// +// Under delta temporality an attribute set that disappears for one collection cycle and then +// reappears must emit only the increment since the last observed value, not the full new absolute +// value. The cumulative baseline (cumulative_hash_map_) is preserved across absent cycles so that +// the delta computation in Record() remains correct. +TEST(AsyncMetricStorageRegressionTest, AttributeReappearanceAfterGapDeltaTemporality) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + // Some computation here + auto collection_ts = sdk_start_ts + std::chrono::seconds(5); + + std::shared_ptr collector( + new MockCollectorHandle(AggregationTemporality::kDelta)); + std::vector> collectors; + collectors.push_back(collector); + + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + // Collection 1: A=10 → delta should be 10. + std::unordered_map measurements1 = { + {{{"attr", "A"}}, 10}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int64_t delta_value = -1; + storage.Collect(collector.get(), collectors, sdk_start_ts, collection_ts, + [&](const MetricData &metric_data) { + for (const auto &data_attr : metric_data.point_data_attr_) + { + delta_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + EXPECT_EQ(delta_value, 10); + + // Collection 2: attribute A absent – nothing recorded, nothing emitted. + std::unordered_map measurements2; + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int attr_count = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, + collection_ts + std::chrono::seconds(5), [&](const MetricData &metric_data) { + attr_count += static_cast(metric_data.point_data_attr_.size()); + return true; + }); + EXPECT_EQ(attr_count, 0) << "No data points expected when attribute set is absent"; + + // Collection 3: A reappears with absolute value 11. + // The cumulative baseline (10) was preserved across the absent cycle, so + // delta = 11 - 10 = 1 — the correct increment since the attribute was last seen. + std::unordered_map measurements3 = { + {{{"attr", "A"}}, 11}}; + storage.RecordLong(measurements3, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + delta_value = -1; + storage.Collect(collector.get(), collectors, sdk_start_ts, + collection_ts + std::chrono::seconds(10), [&](const MetricData &metric_data) { + for (const auto &data_attr : metric_data.point_data_attr_) + { + delta_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + // Baseline was preserved: delta = new_value - last_seen_value = 11 - 10 = 1. + EXPECT_EQ(delta_value, 1) + << "After a gap, reappearing attribute must emit the increment since last seen"; +} + } // namespace