Remove stale attribute combinations that are no longer observed in ObservableCounter and ObservableUpDownCounter

Author: taisho6339

opentelemetry-sdk/CHANGELOG.md

@@ -5,6 +5,10 @@
 - Added `Resource::get_ref(&self, key: &Key) -> Option<&Value>` to allow retrieving a reference to a resource value without cloning.
 - **Breaking** Removed the following public hidden methods from the `SdkTracer` [#3227][3227]:
   - `id_generator`, `should_sample`
+- **Fix**: ObservableCounter and ObservableUpDownCounter now correctly report only data points from the current measurement cycle,
+  removing stale attribute combinations that are no longer observed. This fixes [#3213](3213)
+  where dynamic attributes (e.g., client connections with changing IPs/ports) would accumulate historical data indefinitely.
+  Both Delta and Cumulative temporality modes now use `collect_and_reset` internally to clean up unobserved attributes.
 
 [3227]: https://github.com/open-telemetry/opentelemetry-rust/pull/3227
 

opentelemetry-sdk/src/metrics/internal/precomputed_sum.rs

@@ -116,8 +116,10 @@ impl<T: Number> PrecomputedSum<T> {
         s_data.temporality = Temporality::Cumulative;
         s_data.is_monotonic = self.monotonic;
 
+        // Use collect_and_reset to remove stale attributes (not observed in current callback)
+        // For cumulative, report absolute values (no delta calculation needed)
         self.value_map
-            .collect_readonly(&mut s_data.data_points, |attributes, aggr| SumDataPoint {
+            .collect_and_reset(&mut s_data.data_points, |attributes, aggr| SumDataPoint {
                 attributes,
                 value: aggr.value.get_value(),
                 exemplars: vec![],

opentelemetry-sdk/src/metrics/mod.rs

@@ -1460,14 +1460,12 @@ mod tests {
         asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
             "gauge", true,
         );
-        // TODO fix: all asynchronous instruments should not emit data points if not measured
-        // but these implementations are still buggy
         asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
-            "counter", false,
+            "counter", true,
         );
         asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
             "updown_counter",
-            false,
+            true,
         );
     }
 

opentelemetry-sdk/src/metrics/pipeline.rs

@@ -524,9 +524,8 @@ fn aggregate_fn<T: Number>(
         }
         Aggregation::Sum => {
             let fns = match kind {
-                // TODO implement: observable instruments should not report data points on every collect
-                // from SDK: For asynchronous instruments with Delta or Cumulative aggregation temporality,
-                // MetricReader.Collect MUST only receive data points with measurements recorded since the previous collection
+                // Observable instruments use collect_and_reset to report only data points
+                // measured in the current callback, removing stale attributes
                 InstrumentKind::ObservableCounter => b.precomputed_sum(true),
                 InstrumentKind::ObservableUpDownCounter => b.precomputed_sum(false),
                 InstrumentKind::Counter | InstrumentKind::Histogram => b.sum(true),