[METRICS] Fix stale async attribute sets in cumulative exports (#4108)

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

sdk/include/opentelemetry/sdk/metrics/state/async_metric_storage.h

@@ -8,6 +8,7 @@
 #include <mutex>
 #include <unordered_map>
 #include <utility>
+#include <vector>
 
 #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<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
         delta_hash_map_(
             std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
         round_hash_map_(
@@ -135,9 +134,13 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
     std::shared_ptr<AttributesHashMap> delta_metrics = nullptr;
     {
       std::lock_guard<std::mutex> 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 <class T>
   void RecordMonotonicSum(const MetricAttributes &attributes, T value)
   {
-    auto observed = create_default_aggregation_();
-    observed->Aggregate(value);
-
-    std::unique_ptr<Aggregation> 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<Aggregation> 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<AttributesHashMap>(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<std::unique_ptr<Aggregation>()> create_default_aggregation_;
 
-  std::unique_ptr<AttributesHashMap> last_observed_hash_map_;
+  std::unordered_map<CollectorHandle *, std::unique_ptr<AttributesHashMap>>
+      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<MetricAttributes, std::unique_ptr<Aggregation>, AttributeHashGenerator>
+      monotonic_round_observations_;
+  std::vector<MetricAttributes> monotonic_round_attribute_order_;
+  std::unordered_map<
+      CollectorHandle *,
+      std::unordered_map<MetricAttributes, std::unique_ptr<Aggregation>, AttributeHashGenerator>>
+      last_observed_monotonic_hash_maps_;
+#ifdef UNIT_TESTING
+  friend class AsyncMetricStorageTestPeer;
+#endif
   std::unique_ptr<AttributesHashMap> delta_hash_map_;
   std::unique_ptr<AttributesHashMap> round_hash_map_;
 

sdk/src/metrics/state/temporal_metric_storage.cc

@@ -48,6 +48,8 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,
   std::lock_guard<std::mutex> 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<AttributesHashMap> 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<AttributesHashMap> 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);
 }