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..8578599fe2 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 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 50fa1a2240..a752c7430e 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 71af45ab0b..d57908d694 100644 --- a/sdk/src/metrics/state/temporal_metric_storage.cc +++ b/sdk/src/metrics/state/temporal_metric_storage.cc @@ -6,6 +6,7 @@ #include #include #include +#include #include #include @@ -31,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, @@ -135,6 +138,23 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, return true; }); } + + // For async cumulative exports, capture the attribute sets actually observed for THIS collector + // during THIS cycle (the freshly merged deltas), before the cumulative baseline is merged in. + // delta_metrics only carries the deltas for whichever collector drains the shared map first, so + // it cannot be used as the "observed this cycle" signal in multi-collector setups. + const bool async_cumulative = + is_async_ && aggregation_temporarily == AggregationTemporality::kCumulative; + std::unordered_set observed_this_cycle; + if (async_cumulative) + { + merged_metrics->GetAllEntries( + [&observed_this_cycle](const MetricAttributes &attributes, Aggregation &) { + observed_this_cycle.insert(attributes); + return true; + }); + } + // Get the last reported metrics for the `collector` from `last reported metrics` stash // - If the aggregation_temporarily for the collector is cumulative // - Merge the last reported metrics with unreported metrics (which is in merged_metrics), @@ -160,6 +180,9 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, } else { + // Always carry forward cumulative state so the baseline survives gaps. + // For async instruments, stale suppression is applied only when building + // the exported MetricData below. auto def_agg = DefaultAggregation::CreateAggregation( aggregation_type_, instrument_descriptor_, aggregation_config_); merged_metrics->Set(attributes, def_agg->Merge(aggregation)); @@ -193,13 +216,26 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, metric_data.start_ts = last_collection_ts; metric_data.end_ts = collection_ts; result_to_export->GetAllEntries( - [&metric_data](const MetricAttributes &attributes, Aggregation &aggregation) { + [&metric_data, &observed_this_cycle, async_cumulative](const MetricAttributes &attributes, + Aggregation &aggregation) { + if (async_cumulative && observed_this_cycle.find(attributes) == observed_this_cycle.end()) + { + // Async cumulative exports must omit attribute sets that were not observed + // in the current callback cycle, while keeping the internal cumulative state. + return true; + } + PointDataAttributes point_data_attr; point_data_attr.point_data = aggregation.ToPoint(); point_data_attr.attributes = attributes; metric_data.point_data_attr_.emplace_back(std::move(point_data_attr)); return true; }); + + if (metric_data.point_data_attr_.empty()) + { + return true; + } return callback(metric_data); } diff --git a/sdk/test/metrics/async_metric_storage_test.cc b/sdk/test/metrics/async_metric_storage_test.cc index 54eec782ec..e3dcc9cf17 100644 --- a/sdk/test/metrics/async_metric_storage_test.cc +++ b/sdk/test/metrics/async_metric_storage_test.cc @@ -52,6 +52,10 @@ class WritableMetricStorageTestObservableGaugeFixture : public ::testing::TestWithParam {}; +class AsyncMetricStorageStaleAttributeFixture + : public ::testing::TestWithParam +{}; + TEST_P(AsyncWritableMetricStorageTestFixture, TestAggregation) { AggregationTemporality temporality = GetParam(); @@ -322,4 +326,451 @@ INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestObservableGaugeFixtureLong, ::testing::Values(AggregationTemporality::kCumulative, AggregationTemporality::kDelta)); +// Async instruments must NOT carry forward attribute sets that were not reported by the callback in +// the current collection cycle. +TEST_P(AsyncMetricStorageStaleAttributeFixture, StaleAttributeSetDropped) +{ + const AggregationTemporality temporality = GetParam(); + + 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(temporality)); + 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"; + EXPECT_EQ(get_count, 1); + // Cumulative exports the absolute value (20); delta exports the increment since last seen (10). + const int64_t expected_get_value = (temporality == AggregationTemporality::kCumulative) ? 20 : 10; + EXPECT_EQ(get_value, expected_get_value); +} + +// An attribute set that disappears for one collection cycle and then reappears must preserve the +// cumulative baseline across the absent cycle. On reappearance cumulative exports the full absolute +// value, while delta exports only the increment since the attribute was last seen. +TEST_P(AsyncMetricStorageStaleAttributeFixture, AttributeReappearanceAfterGap) +{ + const AggregationTemporality temporality = GetParam(); + + 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(temporality)); + 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 -> both temporalities export 10. + std::unordered_map measurements1 = { + {{{"attr", "A"}}, 10}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int64_t 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_) + { + value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + EXPECT_EQ(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 30. + std::unordered_map measurements3 = { + {{{"attr", "A"}}, 30}}; + storage.RecordLong(measurements3, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + 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_) + { + value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + // Baseline (10) preserved across the gap: cumulative = 30, delta = 30 - 10 = 20. + const int64_t expected_value = (temporality == AggregationTemporality::kCumulative) ? 30 : 20; + EXPECT_EQ(value, expected_value) + << "After a gap, reappearing attribute must preserve the baseline across the absent cycle"; +} + +// Stale suppression must apply to every collector, not just the one that drains the shared delta +// first. With two collectors, an attribute dropped by the callback must disappear from both +// collectors' exports while the still-reported attribute keeps its correct value for both. +TEST_P(AsyncMetricStorageStaleAttributeFixture, StaleAttributeSetDroppedMultiCollector) +{ + const AggregationTemporality temporality = GetParam(); + + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + auto collection_ts1 = sdk_start_ts + std::chrono::seconds(5); + auto collection_ts2 = sdk_start_ts + std::chrono::seconds(10); + + std::shared_ptr collector1(new MockCollectorHandle(temporality)); + std::shared_ptr collector2(new MockCollectorHandle(temporality)); + std::vector> collectors; + collectors.push_back(collector1); + collectors.push_back(collector2); + + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + auto collect_request_types = [&](CollectorHandle *collector, + opentelemetry::common::SystemTimestamp collection_ts, + int &get_count, int &put_count, int64_t &get_value) { + get_count = 0; + put_count = 0; + get_value = 0; + storage.Collect(collector, 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++; + get_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + else if (key == "PUT") + { + put_count++; + } + } + return true; + }); + }; + + // Collection 1: both GET and PUT reported – both collectors see both attribute sets. + std::unordered_map measurements1 = { + {{{"RequestType", "GET"}}, 10}, {{{"RequestType", "PUT"}}, 5}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int c1_get = 0, c1_put = 0, c2_get = 0, c2_put = 0; + int64_t c1_get_value = 0, c2_get_value = 0; + collect_request_types(collector1.get(), collection_ts1, c1_get, c1_put, c1_get_value); + collect_request_types(collector2.get(), collection_ts1, c2_get, c2_put, c2_get_value); + EXPECT_EQ(c1_get, 1); + EXPECT_EQ(c1_put, 1); + EXPECT_EQ(c2_get, 1); + EXPECT_EQ(c2_put, 1) << "Second collector must observe PUT even though the first drained the " + "shared delta"; + + // Collection 2: only GET reported – PUT is dropped by the callback. + std::unordered_map measurements2 = { + {{{"RequestType", "GET"}}, 20}}; + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + collect_request_types(collector1.get(), collection_ts2, c1_get, c1_put, c1_get_value); + collect_request_types(collector2.get(), collection_ts2, c2_get, c2_put, c2_get_value); + + // Cumulative exports the absolute value (20); delta exports the increment since last seen (10). + const int64_t expected_get_value = (temporality == AggregationTemporality::kCumulative) ? 20 : 10; + + EXPECT_EQ(c1_put, 0) << "Stale PUT must be dropped for the first collector"; + EXPECT_EQ(c2_put, 0) << "Stale PUT must be dropped for the second collector too"; + EXPECT_EQ(c1_get, 1); + EXPECT_EQ(c2_get, 1); + EXPECT_EQ(c1_get_value, expected_get_value); + EXPECT_EQ(c2_get_value, expected_get_value) + << "Second collector must independently receive the still-reported GET value"; +} + +// A collector that skips a collection cycle must not lose values observed during its interval. +// Deltas drained by another collector are stashed for every collector, so when the lagging +// collector finally collects it receives the accumulated value — even though delta_metrics is empty +// at that moment (already drained by the other collector). This is the exact case the per-collector +// observed set fixes: suppression is driven by the collector's own unreported deltas, not by +// delta_metrics. +TEST_P(AsyncMetricStorageStaleAttributeFixture, MultiCollectorLaggingCollector) +{ + const AggregationTemporality temporality = GetParam(); + + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + auto collection_ts1 = sdk_start_ts + std::chrono::seconds(5); + auto collection_ts2 = sdk_start_ts + std::chrono::seconds(10); + + std::shared_ptr collector1(new MockCollectorHandle(temporality)); + std::shared_ptr collector2(new MockCollectorHandle(temporality)); + std::vector> collectors; + collectors.push_back(collector1); + collectors.push_back(collector2); + + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + auto collect = [&](CollectorHandle *collector, + opentelemetry::common::SystemTimestamp collection_ts, int &count, + int64_t &value_out) { + count = 0; + value_out = 0; + storage.Collect(collector, collectors, sdk_start_ts, collection_ts, + [&](const MetricData &metric_data) { + for (const auto &data_attr : metric_data.point_data_attr_) + { + count++; + value_out = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + }; + + int c1_count = 0, c2_count = 0; + int64_t c1_value = 0, c2_value = 0; + + // Cycle 1: A=10 observed. Only collector1 collects; collector2 lags behind. + std::unordered_map measurements1 = { + {{{"attr", "A"}}, 10}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + collect(collector1.get(), collection_ts1, c1_count, c1_value); + EXPECT_EQ(c1_count, 1); + EXPECT_EQ(c1_value, 10); + + // Cycle 2: A=20 observed. Only collector1 collects again; collector2 still lags. + std::unordered_map measurements2 = { + {{{"attr", "A"}}, 20}}; + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + collect(collector1.get(), collection_ts2, c1_count, c1_value); + EXPECT_EQ(c1_count, 1); + // Cumulative: absolute 20; delta: increment since previous collection (20 - 10 = 10). + EXPECT_EQ(c1_value, (temporality == AggregationTemporality::kCumulative) ? 20 : 10); + + // collector2 finally collects. delta_metrics is empty now (collector1 already drained it), but + // collector2's stash accumulated both cycles -> it must still receive the value. + collect(collector2.get(), collection_ts2, c2_count, c2_value); + EXPECT_EQ(c2_count, 1) << "Lagging collector must not drop a value observed during its interval"; + // Both temporalities yield 20 here: cumulative absolute is 20, and delta since collector2's first + // (never) collection is the full accumulated 10 + 10 = 20. + EXPECT_EQ(c2_value, 20) + << "Lagging collector must receive the accumulated value from the cycles it missed"; +} + +// With more than one collector, delta temporality no longer short-circuits on the single-collector +// fast path — it goes through the slow path instead. This exercises the delta branch that sets +// start_ts to the previous collection's timestamp, and verifies both collectors independently +// receive the correct value. Under cumulative every collector must observe the absolute value even +// though only the first collector of a cycle drains the shared delta (the per-collector observed +// set drives stale suppression, not delta_metrics), and start_ts stays at the SDK start. +TEST_P(AsyncMetricStorageStaleAttributeFixture, MultiCollector) +{ + const AggregationTemporality temporality = GetParam(); + + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + auto collection_ts1 = sdk_start_ts + std::chrono::seconds(5); + auto collection_ts2 = sdk_start_ts + std::chrono::seconds(10); + + std::shared_ptr collector1(new MockCollectorHandle(temporality)); + std::shared_ptr collector2(new MockCollectorHandle(temporality)); + std::vector> collectors; + collectors.push_back(collector1); + collectors.push_back(collector2); + + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + auto collect_value = [&](CollectorHandle *collector, + opentelemetry::common::SystemTimestamp collection_ts, int64_t &value_out, + opentelemetry::common::SystemTimestamp &start_ts_out, + opentelemetry::common::SystemTimestamp &end_ts_out) { + storage.Collect(collector, collectors, sdk_start_ts, collection_ts, + [&](const MetricData &metric_data) { + start_ts_out = metric_data.start_ts; + end_ts_out = metric_data.end_ts; + for (const auto &data_attr : metric_data.point_data_attr_) + { + value_out = opentelemetry::nostd::get( + opentelemetry::nostd::get(data_attr.point_data).value_); + } + return true; + }); + }; + + // Cycle 1: A=10 observed once, both collectors drain the same value -> each sees 10. + std::unordered_map measurements1 = { + {{{"attr", "A"}}, 10}}; + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + int64_t c1_value = -1; + int64_t c2_value = -1; + opentelemetry::common::SystemTimestamp c1_start; + opentelemetry::common::SystemTimestamp c2_start; + opentelemetry::common::SystemTimestamp c1_end; + opentelemetry::common::SystemTimestamp c2_end; + collect_value(collector1.get(), collection_ts1, c1_value, c1_start, c1_end); + collect_value(collector2.get(), collection_ts1, c2_value, c2_start, c2_end); + EXPECT_EQ(c1_value, 10); + EXPECT_EQ(c2_value, 10) << "Second collector must observe the value even though the first " + "collector drained the shared delta"; + EXPECT_EQ(c1_end, opentelemetry::common::SystemTimestamp(collection_ts1)) + << "end_ts must equal the collection timestamp"; + EXPECT_EQ(c2_end, opentelemetry::common::SystemTimestamp(collection_ts1)) + << "end_ts must equal the collection timestamp"; + + // Cycle 2: A=30 observed once. + std::unordered_map measurements2 = { + {{{"attr", "A"}}, 30}}; + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + c1_value = -1; + c2_value = -1; + collect_value(collector1.get(), collection_ts2, c1_value, c1_start, c1_end); + collect_value(collector2.get(), collection_ts2, c2_value, c2_start, c2_end); + + // end_ts always advances to the current collection timestamp, for both temporalities. + EXPECT_EQ(c1_end, opentelemetry::common::SystemTimestamp(collection_ts2)) + << "end_ts must equal the current collection timestamp"; + EXPECT_EQ(c2_end, opentelemetry::common::SystemTimestamp(collection_ts2)) + << "end_ts must equal the current collection timestamp"; + + if (temporality == AggregationTemporality::kCumulative) + { + // Cumulative reports the absolute value and keeps start_ts at the SDK start. + EXPECT_EQ(c1_value, 30); + EXPECT_EQ(c2_value, 30) << "Second collector must independently receive the absolute value"; + EXPECT_EQ(c1_start, opentelemetry::common::SystemTimestamp(sdk_start_ts)) + << "Cumulative start_ts must stay at the SDK start"; + EXPECT_EQ(c2_start, opentelemetry::common::SystemTimestamp(sdk_start_ts)) + << "Cumulative start_ts must stay at the SDK start"; + } + else + { + // Delta reports the increment since each collector's previous collection. + EXPECT_EQ(c1_value, 20) << "Delta since previous collection must be 30 - 10 = 20"; + EXPECT_EQ(c2_value, 20) << "Second collector must independently receive the same increment"; + // The slow-path delta branch sets start_ts to the previous collection's timestamp. + EXPECT_EQ(c1_start, opentelemetry::common::SystemTimestamp(collection_ts1)) + << "Delta start_ts must continue from the previous collection"; + EXPECT_EQ(c2_start, opentelemetry::common::SystemTimestamp(collection_ts1)) + << "Delta start_ts must continue from the previous collection"; + } +} + +INSTANTIATE_TEST_SUITE_P(AsyncMetricStorageRegression, + AsyncMetricStorageStaleAttributeFixture, + ::testing::Values(AggregationTemporality::kCumulative, + AggregationTemporality::kDelta)); + } // namespace