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 b1c31cf32..c2fc16061 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 50fa1a224..a752c7430 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 71af45ab0..7bd87afe2 100644 --- a/sdk/src/metrics/state/temporal_metric_storage.cc +++ b/sdk/src/metrics/state/temporal_metric_storage.cc @@ -31,11 +31,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, @@ -158,8 +160,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 54eec782e..6e7ce2275 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