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

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 <class T>
@@ -134,6 +134,12 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
       delta_metrics = std::move(delta_hash_map_);
       delta_hash_map_ =
           std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_);
+      // cumulative_hash_map_ is intentionally NOT pruned here.
+      // It preserves the last-seen absolute value for every attribute set so that
+      // delta computation in Record() remains correct if an attribute set reappears
+      // after being absent for one or more collection cycles.
+      // Stale entries are suppressed at export time by the is_async_ guard in
+      // TemporalMetricStorage::buildMetrics() instead.
     }
 
     auto status =

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<std::shared_ptr<CollectorHandle>> 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

sdk/src/metrics/state/temporal_metric_storage.cc

@@ -31,11 +31,13 @@ namespace metrics
 
 TemporalMetricStorage::TemporalMetricStorage(InstrumentDescriptor instrument_descriptor,
                                              AggregationType aggregation_type,
-                                             const AggregationConfig *aggregation_config)
+                                             const AggregationConfig *aggregation_config,
+                                             bool is_async)
     : instrument_descriptor_(std::move(instrument_descriptor)),
       aggregation_type_(aggregation_type),
       aggregation_config_(aggregation_config),
-      instrument_creation_ts_(std::chrono::system_clock::now())
+      instrument_creation_ts_(std::chrono::system_clock::now()),
+      is_async_(is_async)
 {}
 
 bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,
@@ -158,8 +160,11 @@ bool TemporalMetricStorage::buildMetrics(CollectorHandle *collector,
             {
               merged_metrics->Set(attributes, agg->Merge(aggregation));
             }
-            else
+            else if (!is_async_)
             {
+              // For sync instruments, carry forward attribute sets not observed this cycle.
+              // For async instruments, drop them per the spec: the SDK SHOULD NOT produce
+              // aggregated metric data for attribute sets not observed in the current callback.
               auto def_agg = DefaultAggregation::CreateAggregation(
                   aggregation_type_, instrument_descriptor_, aggregation_config_);
               merged_metrics->Set(attributes, def_agg->Merge(aggregation));

sdk/test/metrics/async_metric_storage_test.cc

@@ -322,4 +322,169 @@ INSTANTIATE_TEST_SUITE_P(WritableMetricStorageTestObservableGaugeFixtureLong,
                          ::testing::Values(AggregationTemporality::kCumulative,
                                            AggregationTemporality::kDelta));
 
+// Regression test for https://github.com/open-telemetry/opentelemetry-cpp/issues/4108
+//
+// Async instruments under cumulative temporality must NOT carry forward attribute sets that were
+// not reported by the callback in the current collection cycle.
+TEST(AsyncMetricStorageRegressionTest, StaleAttributeSetDroppedInCumulativeExport)
+{
+  InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter,
+                                     InstrumentValueType::kLong};
+
+  auto sdk_start_ts = std::chrono::system_clock::now();
+  // Some computation here
+  auto collection_ts = sdk_start_ts + std::chrono::seconds(5);
+
+  std::shared_ptr<CollectorHandle> collector(
+      new MockCollectorHandle(AggregationTemporality::kCumulative));
+  std::vector<std::shared_ptr<CollectorHandle>> 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<MetricAttributes, int64_t, AttributeHashGenerator> 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<std::string>(
+                          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<MetricAttributes, int64_t, AttributeHashGenerator> 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<std::string>(
+                          data_attr.attributes.find("RequestType")->second);
+                      if (key == "GET")
+                      {
+                        get_count++;
+                        get_value = opentelemetry::nostd::get<int64_t>(
+                            opentelemetry::nostd::get<SumPointData>(data_attr.point_data).value_);
+                      }
+                      else if (key == "PUT")
+                      {
+                        put_count++;
+                      }
+                    }
+                    return true;
+                  });
+
+  // PUT must not appear – it was absent from the callback this cycle.
+  EXPECT_EQ(put_count, 0) << "Stale PUT attribute set must be dropped from cumulative export";
+  EXPECT_EQ(get_count, 1);
+  EXPECT_EQ(get_value, 20);
+}
+
+// Regression test for https://github.com/open-telemetry/opentelemetry-cpp/issues/4108
+//
+// Under delta temporality an attribute set that disappears for one collection cycle and then
+// reappears must emit only the increment since the last observed value, not the full new absolute
+// value. The cumulative baseline (cumulative_hash_map_) is preserved across absent cycles so that
+// the delta computation in Record() remains correct.
+TEST(AsyncMetricStorageRegressionTest, AttributeReappearanceAfterGapDeltaTemporality)
+{
+  InstrumentDescriptor instr_desc = {"name", "desc", "1unit", InstrumentType::kObservableCounter,
+                                     InstrumentValueType::kLong};
+
+  auto sdk_start_ts = std::chrono::system_clock::now();
+  // Some computation here
+  auto collection_ts = sdk_start_ts + std::chrono::seconds(5);
+
+  std::shared_ptr<CollectorHandle> collector(
+      new MockCollectorHandle(AggregationTemporality::kDelta));
+  std::vector<std::shared_ptr<CollectorHandle>> collectors;
+  collectors.push_back(collector);
+
+  opentelemetry::sdk::metrics::AsyncMetricStorage storage(
+      instr_desc, AggregationType::kSum,
+#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
+      ExemplarFilterType::kAlwaysOff, ExemplarReservoir::GetNoExemplarReservoir(),
+#endif
+      nullptr);
+
+  // Collection 1: A=10 → delta should be 10.
+  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements1 = {
+      {{{"attr", "A"}}, 10}};
+  storage.RecordLong(measurements1,
+                     opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
+
+  int64_t delta_value = -1;
+  storage.Collect(collector.get(), collectors, sdk_start_ts, collection_ts,
+                  [&](const MetricData &metric_data) {
+                    for (const auto &data_attr : metric_data.point_data_attr_)
+                    {
+                      delta_value = opentelemetry::nostd::get<int64_t>(
+                          opentelemetry::nostd::get<SumPointData>(data_attr.point_data).value_);
+                    }
+                    return true;
+                  });
+  EXPECT_EQ(delta_value, 10);
+
+  // Collection 2: attribute A absent – nothing recorded, nothing emitted.
+  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> 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<int>(metric_data.point_data_attr_.size());
+                    return true;
+                  });
+  EXPECT_EQ(attr_count, 0) << "No data points expected when attribute set is absent";
+
+  // Collection 3: A reappears with absolute value 11.
+  // The cumulative baseline (10) was preserved across the absent cycle, so
+  // delta = 11 - 10 = 1 — the correct increment since the attribute was last seen.
+  std::unordered_map<MetricAttributes, int64_t, AttributeHashGenerator> measurements3 = {
+      {{{"attr", "A"}}, 11}};
+  storage.RecordLong(measurements3,
+                     opentelemetry::common::SystemTimestamp(std::chrono::system_clock::now()));
+
+  delta_value = -1;
+  storage.Collect(collector.get(), collectors, sdk_start_ts,
+                  collection_ts + std::chrono::seconds(10), [&](const MetricData &metric_data) {
+                    for (const auto &data_attr : metric_data.point_data_attr_)
+                    {
+                      delta_value = opentelemetry::nostd::get<int64_t>(
+                          opentelemetry::nostd::get<SumPointData>(data_attr.point_data).value_);
+                    }
+                    return true;
+                  });
+  // Baseline was preserved: delta = new_value - last_seen_value = 11 - 10 = 1.
+  EXPECT_EQ(delta_value, 1)
+      << "After a gap, reappearing attribute must emit the increment since last seen";
+}
+
 }  // namespace