diff --git a/CHANGELOG.md b/CHANGELOG.md index a0bde603a7..4f9e637afd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,16 @@ Increment the: ## [Unreleased] +* [METRICS SDK] Stop exporting stale async attribute sets with cumulative + temporality. Preserve reader-scoped callback behavior and overflow + contributions, and bound per-reader monotonic counter baseline storage by + the configured cardinality limit. + [#4484](https://github.com/open-telemetry/opentelemetry-cpp/pull/4484) + +* [SDK] Preserve the default unlimited timeout when batch span and log record + processors shut down their exporters. + [#4686](https://github.com/open-telemetry/opentelemetry-cpp/pull/4686) + * [API/SDK] Add `Enabled()` to synchronous metric instruments (ABI v2). It returns `false` when the Meter is disabled or every matching view uses the Drop aggregation diff --git a/sdk/include/opentelemetry/sdk/metrics/instruments.h b/sdk/include/opentelemetry/sdk/metrics/instruments.h index a41ffa5d77..3ef9e12cb7 100644 --- a/sdk/include/opentelemetry/sdk/metrics/instruments.h +++ b/sdk/include/opentelemetry/sdk/metrics/instruments.h @@ -155,6 +155,24 @@ struct InstrumentDescriptorUtil return "Unknown"; } } + + static bool IsInstrumentTypeAsync(InstrumentType type) noexcept + { + switch (type) + { + case InstrumentType::kCounter: + case InstrumentType::kGauge: + case InstrumentType::kHistogram: + case InstrumentType::kUpDownCounter: + return false; + case InstrumentType::kObservableCounter: + case InstrumentType::kObservableGauge: + case InstrumentType::kObservableUpDownCounter: + return true; + default: + return false; + } + } }; struct AggregationUtil 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 72d14f378e..6c10cf622f 100644 --- a/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h +++ b/sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h @@ -8,6 +8,7 @@ #include #include #include +#include #include "opentelemetry/nostd/shared_ptr.h" #include "opentelemetry/sdk/common/attributemap_hash.h" @@ -51,8 +52,6 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora aggregation_config_{AggregationConfig::GetOrDefault(aggregation_config)}, attributes_processor_{std::move(attributes_processor)}, is_monotonic_sum_{IsMonotonicSum(aggregation_type, instrument_descriptor)}, - last_observed_hash_map_( - std::make_unique(aggregation_config_->cardinality_limit_)), delta_hash_map_( std::make_unique(aggregation_config_->cardinality_limit_)), round_hash_map_( @@ -135,9 +134,13 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora std::shared_ptr delta_metrics = nullptr; { std::lock_guard guard(hashmap_lock_); - if (!is_monotonic_sum_) + if (is_monotonic_sum_) { - BuildDeltaFromRound(); + BuildMonotonicDeltaFromRound(collector); + } + else + { + BuildDeltaFromRound(collector); } delta_metrics = std::move(delta_hash_map_); delta_hash_map_ = @@ -159,23 +162,14 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora template void RecordMonotonicSum(const MetricAttributes &attributes, T value) { - auto observed = create_default_aggregation_(); - observed->Aggregate(value); - - std::unique_ptr delta; - auto previous = last_observed_hash_map_->Get(attributes); - if (previous) + auto observed = monotonic_round_observations_.find(attributes); + if (observed == monotonic_round_observations_.end()) { - delta = previous->Diff(*observed); + observed = + monotonic_round_observations_.emplace(attributes, create_default_aggregation_()).first; + monotonic_round_attribute_order_.push_back(attributes); } - else - { - delta = DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, - *observed); - } - last_observed_hash_map_->Set(attributes, std::move(observed)); - - AccumulateDelta(FilterAttributes(attributes), *delta); + observed->second->Aggregate(value); } /** @@ -196,13 +190,56 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora * monotonic sums. An attribute set missing from this round keeps its baseline and contributes * no delta; suppressing the stale output is the temporal storage's job. */ - void BuildDeltaFromRound() noexcept + void BuildMonotonicDeltaFromRound(CollectorHandle *collector) noexcept + { + auto &last_observed = last_observed_monotonic_hash_maps_[collector]; + for (const auto &attributes : monotonic_round_attribute_order_) + { + auto observed = monotonic_round_observations_.find(attributes); + if (observed == monotonic_round_observations_.end()) + { + continue; + } + + std::unique_ptr delta; + auto previous = last_observed.find(attributes); + if (previous != last_observed.end()) + { + delta = previous->second->Diff(*observed->second); + previous->second = DefaultAggregation::CloneAggregation( + aggregation_type_, instrument_descriptor_, *observed->second); + } + else + { + delta = DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, + *observed->second); + // Keep only first-observed baselines up to this reader's cardinality limit. + if (last_observed.size() < aggregation_config_->cardinality_limit_) + { + last_observed.emplace(attributes, + DefaultAggregation::CloneAggregation( + aggregation_type_, instrument_descriptor_, *observed->second)); + } + } + AccumulateDelta(FilterAttributes(attributes), *delta); + } + + monotonic_round_observations_.clear(); + monotonic_round_attribute_order_.clear(); + } + + void BuildDeltaFromRound(CollectorHandle *collector) noexcept { - round_hash_map_->GetAllEntries([this](const MetricAttributes &attributes, - Aggregation &aggregation) { + auto &last_observed = last_observed_hash_maps_[collector]; + if (!last_observed) + { + last_observed = std::make_unique(aggregation_config_->cardinality_limit_); + } + round_hash_map_->GetAllEntries([this, &last_observed](const MetricAttributes &attributes, + Aggregation &aggregation) { auto observed = DefaultAggregation::CloneAggregation(aggregation_type_, instrument_descriptor_, aggregation); - auto previous = last_observed_hash_map_->Get(attributes); + auto previous = last_observed->Get(attributes); if (previous) { delta_hash_map_->Set(attributes, previous->Diff(*observed)); @@ -212,7 +249,7 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora delta_hash_map_->Set(attributes, DefaultAggregation::CloneAggregation( aggregation_type_, instrument_descriptor_, *observed)); } - last_observed_hash_map_->Set(attributes, std::move(observed)); + last_observed->Set(attributes, std::move(observed)); return true; }); @@ -281,7 +318,20 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora bool is_monotonic_sum_; std::function()> create_default_aggregation_; - std::unique_ptr last_observed_hash_map_; + std::unordered_map> + last_observed_hash_maps_; + // Source baselines are reader-specific and must remain distinct beyond the output cardinality + // limit to calculate future per-attribute deltas. + std::unordered_map, AttributeHashGenerator> + monotonic_round_observations_; + std::vector monotonic_round_attribute_order_; + std::unordered_map< + CollectorHandle *, + std::unordered_map, AttributeHashGenerator>> + last_observed_monotonic_hash_maps_; +#ifdef UNIT_TESTING + friend class AsyncMetricStorageTestPeer; +#endif std::unique_ptr delta_hash_map_; std::unique_ptr round_hash_map_; diff --git a/sdk/src/metrics/state/temporal_metric_storage.cc b/sdk/src/metrics/state/temporal_metric_storage.cc index 71af45ab0b..cfc806326f 100644 --- a/sdk/src/metrics/state/temporal_metric_storage.cc +++ b/sdk/src/metrics/state/temporal_metric_storage.cc @@ -48,6 +48,8 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, std::lock_guard guard(lock_); AggregationTemporality aggregation_temporarily = collector->GetAggregationTemporality(instrument_descriptor_.type_); + const bool is_async = + InstrumentDescriptorUtil::IsInstrumentTypeAsync(instrument_descriptor_.type_); // Per OTel spec (issue #4062): the start_ts for the first delta collection // interval must be the instrument creation time, not the MeterProvider // start time (which is what sdk_start_ts carries). For cumulative, sdk_start_ts @@ -95,11 +97,23 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, return callback(metric_data); } + const bool async_cumulative = + is_async && aggregation_temporarily == AggregationTemporality::kCumulative; + bool async_cumulative_overflow_observed = + async_cumulative && delta_metrics->Has(GetOverflowAttributes()); + if (delta_metrics->Size()) { - for (auto &col : collectors) + if (is_async) + { + unreported_metrics_[collector].push_back(delta_metrics); + } + else { - unreported_metrics_[col.get()].push_back(delta_metrics); + for (auto &col : collectors) + { + unreported_metrics_[col.get()].push_back(delta_metrics); + } } } @@ -116,60 +130,69 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, std::unique_ptr merged_metrics( new AttributesHashMap(aggregation_config_ ? aggregation_config_->cardinality_limit_ : kAggregationCardinalityLimit)); - for (auto &agg_hashmap : unreported_list) + auto merge_aggregation = + [this, &merged_metrics, async_cumulative, &async_cumulative_overflow_observed]( + const MetricAttributes &attributes, Aggregation &aggregation, bool is_current_delta) { + auto *existing = merged_metrics->GetOrSetDefault(attributes, [this]() { + return DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_, + aggregation_config_); + }); + if (async_cumulative && is_current_delta && + (attributes == GetOverflowAttributes() || !merged_metrics->Has(attributes))) + { + async_cumulative_overflow_observed = true; + } + auto merged = existing->Merge(aggregation); + const auto &merged_attributes = + merged_metrics->Has(attributes) ? attributes : GetOverflowAttributes(); + merged_metrics->Set(merged_attributes, std::move(merged)); + return true; + }; + + auto reported = last_reported_metrics_.find(collector); + std::unique_ptr last_aggr_hashmap; + if (async_cumulative) { - agg_hashmap->GetAllEntries( - [&merged_metrics, this](const MetricAttributes &attributes, Aggregation &aggregation) { - auto agg = merged_metrics->Get(attributes); - if (agg) - { - merged_metrics->Set(attributes, agg->Merge(aggregation)); - } - else - { - merged_metrics->Set(attributes, - DefaultAggregation::CreateAggregation( - aggregation_type_, instrument_descriptor_, aggregation_config_) - ->Merge(aggregation)); - } + // Reserve this reader's current callback attributes before carrying forward stale series. + delta_metrics->GetAllEntries( + [this, &merged_metrics](const MetricAttributes &attributes, Aggregation &) { + merged_metrics->GetOrSetDefault(attributes, [this]() { + return DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_, + aggregation_config_); + }); 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), - // Final result of merge would be in merged_metrics. - // - Move the final merge to the `last reported metrics` stash. - // - If the aggregation_temporarily is delta - // - Store the unreported metrics for `collector` (which is in merged_mtrics) to - // `last reported metrics` stash. - - auto reported = last_reported_metrics_.find(collector); if (reported != last_reported_metrics_.end()) { - auto last_aggr_hashmap = std::move(last_reported_metrics_[collector].attributes_map); + last_aggr_hashmap = std::move(reported->second.attributes_map); if (aggregation_temporarily == AggregationTemporality::kCumulative) { - // merge current delta to previous cumulative + // Preserve the existing cumulative series before admitting new ones under the limit. last_aggr_hashmap->GetAllEntries( - [&merged_metrics, this](const MetricAttributes &attributes, Aggregation &aggregation) { - auto agg = merged_metrics->Get(attributes); - if (agg) - { - merged_metrics->Set(attributes, agg->Merge(aggregation)); - } - else - { - auto def_agg = DefaultAggregation::CreateAggregation( - aggregation_type_, instrument_descriptor_, aggregation_config_); - merged_metrics->Set(attributes, def_agg->Merge(aggregation)); - } - return true; + [&merge_aggregation](const MetricAttributes &attributes, Aggregation &aggregation) { + return merge_aggregation(attributes, aggregation, false); }); } - else + } + + for (auto &agg_hashmap : unreported_list) + { + const bool is_current_delta = agg_hashmap.get() == delta_metrics.get(); + agg_hashmap->GetAllEntries([&merge_aggregation, is_current_delta]( + const MetricAttributes &attributes, Aggregation &aggregation) { + return merge_aggregation(attributes, aggregation, is_current_delta); + }); + } + + // Synchronous cumulative storage preserves prior series' cardinality slots. Async cumulative + // storage reserves the current callback's slots above before carrying forward prior state. + + if (reported != last_reported_metrics_.end()) + { + if (aggregation_temporarily != AggregationTemporality::kCumulative) { - last_collection_ts = last_reported_metrics_[collector].collection_ts; + last_collection_ts = reported->second.collection_ts; } last_reported_metrics_[collector] = LastReportedMetrics{std::move(merged_metrics), collection_ts}; @@ -192,14 +215,29 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector, metric_data.aggregation_temporality = aggregation_temporarily; metric_data.start_ts = last_collection_ts; metric_data.end_ts = collection_ts; + // Async observations apply only to the reader whose callback produced delta_metrics. result_to_export->GetAllEntries( - [&metric_data](const MetricAttributes &attributes, Aggregation &aggregation) { + [&metric_data, delta_metrics, async_cumulative, async_cumulative_overflow_observed]( + const MetricAttributes &attributes, Aggregation &aggregation) { + if (async_cumulative && !delta_metrics->Has(attributes) && + !(attributes == GetOverflowAttributes() && async_cumulative_overflow_observed)) + { + // Omit attribute sets not observed by this reader's callback, 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 a11a75a104..c167540b6b 100644 --- a/sdk/test/metrics/async_metric_storage_test.cc +++ b/sdk/test/metrics/async_metric_storage_test.cc @@ -5,8 +5,10 @@ #include #include #include +#include #include #include +#include #include #include #include @@ -30,12 +32,35 @@ #include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h" #include "opentelemetry/sdk/metrics/state/metric_collector.h" #include "opentelemetry/sdk/metrics/view/attributes_processor.h" +#include "opentelemetry/version.h" #ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW # include "opentelemetry/sdk/metrics/exemplar/filter_type.h" # include "opentelemetry/sdk/metrics/exemplar/reservoir.h" #endif +OPENTELEMETRY_BEGIN_NAMESPACE +namespace sdk +{ +namespace metrics +{ + +class AsyncMetricStorageTestPeer +{ +public: + static size_t GetLastObservedMonotonicCount(AsyncMetricStorage &storage, + CollectorHandle *collector) + { + std::lock_guard guard(storage.hashmap_lock_); + auto it = storage.last_observed_monotonic_hash_maps_.find(collector); + return it == storage.last_observed_monotonic_hash_maps_.end() ? 0 : it->second.size(); + } +}; + +} // namespace metrics +} // namespace sdk +OPENTELEMETRY_END_NAMESPACE + using namespace opentelemetry::sdk::metrics; using namespace opentelemetry::sdk::instrumentationscope; using namespace opentelemetry::sdk::resource; @@ -56,6 +81,10 @@ class WritableMetricStorageTestObservableGaugeFixture : public ::testing::TestWithParam {}; +class AsyncMetricStorageStaleAttributeFixture + : public ::testing::TestWithParam +{}; + TEST_P(AsyncWritableMetricStorageTestFixture, TestAggregation) { AggregationTemporality temporality = GetParam(); @@ -330,6 +359,530 @@ class WritableMetricStorageTestFilteredAttributesFixture : public ::testing::TestWithParam {}; +// 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, nullptr, +#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: GET is reported unchanged (and therefore has a zero delta); PUT is dropped. + std::unordered_map measurements2 = { + {{{"RequestType", "GET"}}, 10}}; + 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); + // The unchanged observation is still exported: cumulative value 10, delta value 0. + const int64_t expected_get_value = (temporality == AggregationTemporality::kCumulative) ? 10 : 0; + 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, nullptr, +#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"; +} + +TEST_P(AsyncMetricStorageStaleAttributeFixture, NonMonotonicInstrumentsDropStaleObservations) +{ + const AggregationTemporality temporality = GetParam(); + + for (const auto instrument_type : + {InstrumentType::kObservableGauge, InstrumentType::kObservableUpDownCounter}) + { + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", instrument_type, + InstrumentValueType::kLong}; + + auto sdk_start_ts = std::chrono::system_clock::now(); + auto collection_ts = sdk_start_ts + std::chrono::seconds(5); + std::shared_ptr collector(new MockCollectorHandle(temporality)); + std::vector> collectors{collector}; + + const auto aggregation_type = instrument_type == InstrumentType::kObservableGauge + ? AggregationType::kLastValue + : AggregationType::kSum; + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, aggregation_type, nullptr, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + nullptr); + + std::unordered_map observation = { + {{{"attr", "A"}}, 10}}; + storage.RecordLong(observation, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + size_t point_count = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, collection_ts, + [&](const MetricData &metric_data) { + point_count += metric_data.point_data_attr_.size(); + return true; + }); + EXPECT_EQ(1U, point_count); + + std::unordered_map no_observations; + storage.RecordLong(no_observations, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + point_count = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, + collection_ts + std::chrono::seconds(5), [&](const MetricData &metric_data) { + point_count += metric_data.point_data_attr_.size(); + return true; + }); + EXPECT_EQ(0U, point_count) << "An absent gauge or up-down counter observation must be omitted"; + + observation.begin()->second = 30; + storage.RecordLong(observation, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + point_count = 0; + int64_t reappeared_value = -1; + storage.Collect( + collector.get(), collectors, sdk_start_ts, collection_ts + std::chrono::seconds(10), + [&](const MetricData &metric_data) { + for (const auto &point : metric_data.point_data_attr_) + { + ++point_count; + if (instrument_type == InstrumentType::kObservableGauge) + { + reappeared_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(point.point_data).value_); + } + else + { + reappeared_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(point.point_data).value_); + } + } + return true; + }); + EXPECT_EQ(1U, point_count) << "A reappearing observation must be exported"; + const int64_t expected_value = instrument_type == InstrumentType::kObservableGauge || + temporality == AggregationTemporality::kCumulative + ? 30 + : 20; + EXPECT_EQ(expected_value, reappeared_value); + } +} + +// 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, nullptr, +#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: GET and PUT are reported. Cumulative readers each get a callback round. + 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); + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + 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) << "PUT must be available to the second reader"; + + // 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); + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + 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 lagging reader only exports its own callback's observation when it next collects. +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, nullptr, +#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's callback runs when it collects, and reports the current absolute value. + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + collect(collector2.get(), collection_ts2, c2_count, c2_value); + EXPECT_EQ(c2_count, 1); + EXPECT_EQ(c2_value, 20); +} + +// With more than one collector, delta temporality goes through the slow path. For cumulative +// temporality, record before each reader collects to model its independent callback round. +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, nullptr, +#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: each reader independently observes A=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); + storage.RecordLong(measurements1, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + collect_value(collector2.get(), collection_ts1, c2_value, c2_start, c2_end); + EXPECT_EQ(c1_value, 10); + EXPECT_EQ(c2_value, 10) << "Second reader must export its own callback observation"; + 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: each reader independently observes A=30. + 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); + storage.RecordLong(measurements2, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + 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)); + // Observations the view collapses onto one attribute set are summed, not overwritten. TEST_P(WritableMetricStorageTestFilteredAttributesFixture, TestAggregation) { @@ -693,4 +1246,267 @@ TEST(WritableMetricStorageCardinalityLimit, OverflowCombinesContributions) EXPECT_GT(overflow_value, 0); } +TEST(WritableMetricStorageCardinalityLimit, + AsyncCumulativeCurrentRoundTakesPrecedenceForLaggingReader) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + + std::shared_ptr collector1( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::shared_ptr collector2( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors{collector1, collector2}; + + constexpr size_t kCardinalityLimit = 2; + AggregationConfig aggregation_config{kCardinalityLimit}; + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, nullptr, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + &aggregation_config); + + const auto sdk_start_ts = std::chrono::system_clock::now(); + std::unordered_map first_reader_round = { + {{{"id", "a"}}, 1}, {{{"id", "b"}}, 2}}; + storage.RecordLong(first_reader_round, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + size_t first_reader_points = 0; + storage.Collect(collector1.get(), collectors, sdk_start_ts, + sdk_start_ts + std::chrono::seconds(5), [&](const MetricData &metric_data) { + first_reader_points = metric_data.point_data_attr_.size(); + return true; + }); + EXPECT_EQ(2U, first_reader_points); + + std::unordered_map second_reader_round = { + {{{"id", "x"}}, 7}}; + storage.RecordLong(second_reader_round, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + + auto expect_current_point = [&](CollectorHandle *collector) { + size_t point_count = 0; + size_t current_count = 0; + int64_t current_value = 0; + storage.Collect(collector, collectors, sdk_start_ts, sdk_start_ts + std::chrono::seconds(10), + [&](const MetricData &metric_data) { + point_count = metric_data.point_data_attr_.size(); + for (const auto &point : metric_data.point_data_attr_) + { + auto id = point.attributes.find("id"); + if (id != point.attributes.end() && + opentelemetry::nostd::get(id->second) == "x") + { + ++current_count; + current_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(point.point_data).value_); + } + } + return true; + }); + EXPECT_EQ(1U, point_count); + EXPECT_EQ(1U, current_count); + EXPECT_EQ(7, current_value); + }; + + expect_current_point(collector2.get()); +} + +TEST(WritableMetricStorageCardinalityLimit, AsyncCumulativeOverflowIsExportedForLaggingReader) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + std::shared_ptr collector1( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::shared_ptr collector2( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors{collector1, collector2}; + constexpr size_t kCardinalityLimit = 2; + AggregationConfig aggregation_config{kCardinalityLimit}; + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, nullptr, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + &aggregation_config); + + std::unordered_map first_round = { + {{{"id", "a"}}, 1}, {{{"id", "b"}}, 2}}; + storage.RecordLong(first_round, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + storage.Collect(collector1.get(), collectors, std::chrono::system_clock::now(), + std::chrono::system_clock::now() + std::chrono::seconds(5), + [](const MetricData &) { return true; }); + + for (const auto &measurement : + std::vector>{{"x", 7}, {"y", 8}, {"z", 9}}) + { + std::unordered_map round = { + {{{"id", measurement.first}}, measurement.second}}; + storage.RecordLong(round, + opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + } + + std::map values; + size_t point_count = 0; + storage.Collect( + collector2.get(), collectors, std::chrono::system_clock::now(), + std::chrono::system_clock::now() + std::chrono::seconds(15), + [&](const MetricData &metric_data) { + point_count = metric_data.point_data_attr_.size(); + for (const auto &point : metric_data.point_data_attr_) + { + const auto value = opentelemetry::nostd::get( + opentelemetry::nostd::get(point.point_data).value_); + if (point.attributes.find(kAttributesLimitOverflowKey) != point.attributes.end()) + { + values["overflow"] = value; + } + else + { + values[opentelemetry::nostd::get(point.attributes.find("id")->second)] = + value; + } + } + return true; + }); + + EXPECT_EQ(3U, point_count); + EXPECT_EQ(7, values["x"]); + EXPECT_EQ(8, values["y"]); + EXPECT_EQ(9, values["overflow"]); +} + +TEST(WritableMetricStorageCardinalityLimit, + AsyncCumulativeCurrentRoundTakesPrecedenceOverStaleSeries) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableGauge, + InstrumentValueType::kLong}; + std::shared_ptr collector( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors{collector}; + constexpr size_t kCardinalityLimit = 2; + AggregationConfig aggregation_config{kCardinalityLimit}; + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kLastValue, nullptr, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + &aggregation_config); + + auto record = + [&storage](std::unordered_map &round) { + storage.RecordLong( + round, opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + }; + auto sdk_start_ts = std::chrono::system_clock::now(); + auto collect = [&]() { + std::vector> points; + storage.Collect(collector.get(), collectors, sdk_start_ts, + sdk_start_ts + std::chrono::seconds(5), [&](const MetricData &metric_data) { + for (const auto &point : metric_data.point_data_attr_) + { + const auto &last_value = + opentelemetry::nostd::get(point.point_data); + points.emplace_back(point.attributes, + opentelemetry::nostd::get(last_value.value_)); + } + return true; + }); + return points; + }; + + std::unordered_map first_round = { + {{{"id", "a"}}, 1}, {{{"id", "b"}}, 2}}; + record(first_round); + EXPECT_EQ(2U, collect().size()); + + std::unordered_map second_round = { + {{{"id", "x"}}, 7}}; + record(second_round); + auto points = collect(); + ASSERT_EQ(1U, points.size()); + ASSERT_NE(points[0].first.find("id"), points[0].first.end()); + EXPECT_EQ("x", opentelemetry::nostd::get(points[0].first.find("id")->second)); + EXPECT_EQ(7, points[0].second); + + std::unordered_map third_round = { + {{{"id", "x"}}, 8}, {{{"id", "y"}}, 3}}; + record(third_round); + points = collect(); + ASSERT_EQ(2U, points.size()); + std::map values; + for (const auto &point : points) + { + values[opentelemetry::nostd::get(point.first.find("id")->second)] = point.second; + } + EXPECT_EQ(8, values["x"]); + EXPECT_EQ(3, values["y"]); +} + +TEST_P(AsyncWritableMetricStorageTestFixture, CardinalityLimitCapsCounterBaselines) +{ + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter, + InstrumentValueType::kLong}; + std::shared_ptr collector(new MockCollectorHandle(GetParam())); + std::vector> collectors{collector}; + constexpr size_t kCardinalityLimit = 2; + AggregationConfig aggregation_config{kCardinalityLimit}; + opentelemetry::sdk::metrics::AsyncMetricStorage storage( + instr_desc, AggregationType::kSum, nullptr, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(), +#endif + &aggregation_config); + + auto record = + [&storage](std::unordered_map &round) { + storage.RecordLong( + round, opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now())); + }; + auto collect = [&]() { return CollectSumPoints(storage, collector, collectors); }; + + std::unordered_map first_round = { + {{{"id", "a"}}, 1}, {{{"id", "b"}}, 2}}; + record(first_round); + EXPECT_EQ(2U, collect().size()); + EXPECT_EQ(kCardinalityLimit, + AsyncMetricStorageTestPeer::GetLastObservedMonotonicCount(storage, collector.get())); + + std::unordered_map second_round = { + {{{"id", "x"}}, 7}}; + record(second_round); + auto points = collect(); + ASSERT_EQ(1U, points.size()); + ASSERT_NE(points[0].first.find("id"), points[0].first.end()); + EXPECT_EQ("x", opentelemetry::nostd::get(points[0].first.find("id")->second)); + EXPECT_EQ(7, points[0].second); + EXPECT_EQ(kCardinalityLimit, + AsyncMetricStorageTestPeer::GetLastObservedMonotonicCount(storage, collector.get())); + + std::unordered_map third_round = { + {{{"id", "x"}}, 8}, {{{"id", "y"}}, 3}}; + record(third_round); + points = collect(); + ASSERT_EQ(2U, points.size()); + std::map values; + for (const auto &point : points) + { + values[opentelemetry::nostd::get(point.first.find("id")->second)] = point.second; + } + EXPECT_EQ(GetParam() == AggregationTemporality::kCumulative ? 15 : 8, values["x"]); + EXPECT_EQ(3, values["y"]); + + for (int i = 0; i < 100; ++i) + { + std::unordered_map churn_round = { + {{{"id", std::to_string(i)}}, i + 1}}; + record(churn_round); + collect(); + EXPECT_LE(AsyncMetricStorageTestPeer::GetLastObservedMonotonicCount(storage, collector.get()), + kCardinalityLimit); + } +} + } // namespace diff --git a/sdk/test/metrics/cardinality_limit_test.cc b/sdk/test/metrics/cardinality_limit_test.cc index cab5f91209..99b9dbc281 100644 --- a/sdk/test/metrics/cardinality_limit_test.cc +++ b/sdk/test/metrics/cardinality_limit_test.cc @@ -168,6 +168,70 @@ INSTANTIATE_TEST_SUITE_P(All, WritableMetricStorageCardinalityLimitTestFixture, ::testing::Values(AggregationTemporality::kDelta)); +TEST(CardinalityLimit, SyncCumulativeKeepsExistingSeriesBeforeOverflow) +{ + constexpr size_t kCardinalityLimit = 2; + AggregationConfig aggregation_config{kCardinalityLimit}; + InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kCounter, + InstrumentValueType::kLong}; + std::shared_ptr attributes_processor{ + new DefaultAttributesProcessor{}}; + SyncMetricStorage storage(instr_desc, AggregationType::kSum, attributes_processor, +#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW + ExemplarFilterType::kAlwaysOff, + ExemplarReservoir::GetNoExemplarReservoir(), +#endif + &aggregation_config); + + auto sdk_start_ts = std::chrono::system_clock::now(); + std::shared_ptr collector( + new MockCollectorHandle(AggregationTemporality::kCumulative)); + std::vector> collectors{collector}; + auto record = [&storage](const std::string &id, int64_t value) { + std::map attributes = {{"id", id}}; + storage.RecordLong(value, KeyValueIterableView>(attributes), + opentelemetry::context::Context{}); + }; + + record("a", 10); + record("b", 20); + size_t initial_point_count = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, sdk_start_ts + std::chrono::seconds(5), + [&](const MetricData &metric_data) { + initial_point_count = metric_data.point_data_attr_.size(); + return true; + }); + EXPECT_EQ(2U, initial_point_count); + + record("x", 7); + std::map values; + size_t point_count = 0; + storage.Collect(collector.get(), collectors, sdk_start_ts, + sdk_start_ts + std::chrono::seconds(10), [&](const MetricData &metric_data) { + point_count = metric_data.point_data_attr_.size(); + for (const auto &point : metric_data.point_data_attr_) + { + const auto value = + nostd::get(nostd::get(point.point_data).value_); + auto id = point.attributes.find("id"); + if (id == point.attributes.end()) + { + values["overflow"] = value; + } + else + { + values[nostd::get(id->second)] = value; + } + } + return true; + }); + + EXPECT_EQ(3U, point_count); + EXPECT_EQ(10, values["a"]); + EXPECT_EQ(20, values["b"]); + EXPECT_EQ(7, values["overflow"]); +} + // Pin the overflow attribute key and value to the contract defined in the // OpenTelemetry Metrics SDK specification. The previous key value // `otel.metrics.overflow` (plural) silently diverged from the spec for years diff --git a/sdk/test/metrics/meter_test.cc b/sdk/test/metrics/meter_test.cc index 42c68d3ee9..a6a9e836e2 100644 --- a/sdk/test/metrics/meter_test.cc +++ b/sdk/test/metrics/meter_test.cc @@ -400,6 +400,95 @@ TEST(MeterTest, BasicAsyncTests) observable_counter->RemoveCallback(asyc_generate_measurements_long, nullptr); } +TEST(MeterTest, AsyncCumulativeObservationsAreReaderScoped) +{ + struct CallbackState + { + bool report = true; + size_t round = 0; + } callback_state; + + auto provider = std::make_shared(); + std::shared_ptr reader1(new MockMetricReader()); + std::shared_ptr reader2(new MockMetricReader()); + provider->AddMetricReader(reader1); + provider->AddMetricReader(reader2); + + auto meter = provider->GetMeter("async_reader_scoped_test"); + auto observable_counter = meter->CreateInt64ObservableCounter("request_count"); + auto callback = [](opentelemetry::metrics::ObserverResult observer, void *state) { + auto observer_long = + nostd::get>>(observer); + auto &measurements = *static_cast(state); + ++measurements.round; + if (!measurements.report) + { + return; + } + if (measurements.round % 2 == 1) + { + observer_long->Observe(10, {{"RequestType", "GET"}}); + } + else + { + observer_long->Observe(5, {{"RequestType", "PUT"}}); + } + }; + observable_counter->AddCallback(callback, &callback_state); + + auto collect_and_check = [](const std::shared_ptr &reader, int expected_get_count, + int64_t expected_get_value, int expected_put_count) { + int get_count = 0; + int put_count = 0; + int64_t get_value = 0; + EXPECT_TRUE(reader->Collect([&](ResourceMetrics &resource_metrics) { + for (const auto &scope_metrics : resource_metrics.scope_metric_data_) + { + for (const auto &metric_data : scope_metrics.metric_data_) + { + for (const auto &point : metric_data.point_data_attr_) + { + const auto &request_type = opentelemetry::nostd::get( + point.attributes.find("RequestType")->second); + if (request_type == "GET") + { + get_count++; + get_value = opentelemetry::nostd::get( + opentelemetry::nostd::get(point.point_data).value_); + } + else if (request_type == "PUT") + { + put_count++; + } + } + } + } + return true; + })); + EXPECT_EQ(get_count, expected_get_count); + EXPECT_EQ(put_count, expected_put_count); + if (expected_get_count != 0) + { + EXPECT_EQ(get_value, expected_get_value); + } + }; + + // Each reader runs the callback independently and observes only its own attribute set. + collect_and_check(reader1, 1, 10, 0); + collect_and_check(reader2, 0, 0, 1); + + // Repeating each reader's observation keeps its cumulative value unchanged and scoped. + collect_and_check(reader1, 1, 10, 0); + collect_and_check(reader2, 0, 0, 1); + + // An empty callback round emits no points, even though both readers have history. + callback_state.report = false; + collect_and_check(reader1, 0, 0, 0); + collect_and_check(reader2, 0, 0, 0); + + observable_counter->RemoveCallback(callback, &callback_state); +} + constexpr static unsigned MAX_THREADS = 25; constexpr static unsigned MAX_ITERATIONS_MT = 1000;