Fix aggregation for Gauge (#2021)

Co-authored-by: Cijo Thomas <cijo.thomas@gmail.com>

Author: utpilla

opentelemetry-sdk/CHANGELOG.md

@@ -20,6 +20,9 @@
   ObservableCounter,UpDownCounter including
   [#1517](https://github.com/open-telemetry/opentelemetry-rust/issues/1517).
   [#2004](https://github.com/open-telemetry/opentelemetry-rust/pull/2004)
+- Fixed a bug related to cumulative aggregation of `Gauge` measurements.
+  [#1975](https://github.com/open-telemetry/opentelemetry-rust/issues/1975).
+  [#2021](https://github.com/open-telemetry/opentelemetry-rust/pull/2021)
 
 ## v0.24.1
 

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

@@ -106,13 +106,10 @@ impl<T: Number<T>> AggregateBuilder<T> {
     }
 
     /// Builds a last-value aggregate function input and output.
-    ///
-    /// [Builder::temporality] is ignored and delta is always used.
     pub(crate) fn last_value(&self) -> (impl Measure<T>, impl ComputeAggregation) {
-        // Delta temporality is the only temporality that makes semantic sense for
-        // a last-value aggregate.
         let lv_filter = Arc::new(LastValue::new());
         let lv_agg = Arc::clone(&lv_filter);
+        let t = self.temporality;
 
         (
             self.filter(move |n, a: &[KeyValue]| lv_filter.measure(n, a)),
@@ -127,7 +124,12 @@ impl<T: Number<T>> AggregateBuilder<T> {
                 };
                 let g = g.unwrap_or_else(|| new_agg.as_mut().expect("present if g is none"));
 
-                lv_agg.compute_aggregation(&mut g.data_points);
+                match t {
+                    Some(Temporality::Delta) => {
+                        lv_agg.compute_aggregation_delta(&mut g.data_points)
+                    }
+                    _ => lv_agg.compute_aggregation_cumulative(&mut g.data_points),
+                }
 
                 (g.data_points.len(), new_agg.map(|a| Box::new(a) as Box<_>))
             },

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

@@ -25,7 +25,7 @@ impl<T: Number<T>> LastValue<T> {
         self.value_map.measure(measurement, attrs);
     }
 
-    pub(crate) fn compute_aggregation(&self, dest: &mut Vec<DataPoint<T>>) {
+    pub(crate) fn compute_aggregation_delta(&self, dest: &mut Vec<DataPoint<T>>) {
         let t = SystemTime::now();
         dest.clear();
 
@@ -68,4 +68,48 @@ impl<T: Number<T>> LastValue<T> {
             }
         }
     }
+
+    pub(crate) fn compute_aggregation_cumulative(&self, dest: &mut Vec<DataPoint<T>>) {
+        let t = SystemTime::now();
+        dest.clear();
+
+        // Max number of data points need to account for the special casing
+        // of the no attribute value + overflow attribute.
+        let n = self.value_map.count.load(Ordering::SeqCst) + 2;
+        if n > dest.capacity() {
+            dest.reserve_exact(n - dest.capacity());
+        }
+
+        if self
+            .value_map
+            .has_no_attribute_value
+            .load(Ordering::Acquire)
+        {
+            dest.push(DataPoint {
+                attributes: vec![],
+                start_time: None,
+                time: Some(t),
+                value: self.value_map.no_attribute_tracker.get_value(),
+                exemplars: vec![],
+            });
+        }
+
+        let trackers = match self.value_map.trackers.write() {
+            Ok(v) => v,
+            _ => return,
+        };
+
+        let mut seen = HashSet::new();
+        for (attrs, tracker) in trackers.iter() {
+            if seen.insert(Arc::as_ptr(tracker)) {
+                dest.push(DataPoint {
+                    attributes: attrs.clone(),
+                    start_time: None,
+                    time: Some(t),
+                    value: tracker.get_value(),
+                    exemplars: vec![],
+                });
+            }
+        }
+    }
 }

opentelemetry-sdk/src/metrics/mod.rs

@@ -159,6 +159,7 @@ mod tests {
     use opentelemetry::{metrics::MeterProvider as _, KeyValue};
     use rand::{rngs, Rng, SeedableRng};
     use std::borrow::Cow;
+    use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::{Arc, Mutex};
     use std::thread;
     use std::time::Duration;
@@ -1278,12 +1279,7 @@ mod tests {
         synchronous_instruments_cumulative_with_gap_in_measurements_helper("counter");
         synchronous_instruments_cumulative_with_gap_in_measurements_helper("updown_counter");
         synchronous_instruments_cumulative_with_gap_in_measurements_helper("histogram");
-
-        /* Synchronous Gauge has an aggregation bug. Uncomment the code below to run the test for gauge
-        once this issue is fixed: https://github.com/open-telemetry/opentelemetry-rust/issues/1975
-        */
-
-        // synchronous_instruments_cumulative_with_gap_in_measurements_helper("gauge");
+        synchronous_instruments_cumulative_with_gap_in_measurements_helper("gauge");
     }
 
     fn synchronous_instruments_cumulative_with_gap_in_measurements_helper(
@@ -1404,6 +1400,133 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn asynchronous_instruments_cumulative_with_gap_in_measurements() {
+        // Run this test with stdout enabled to see output.
+        // cargo test asynchronous_instruments_cumulative_with_gap_in_measurements --features=testing -- --nocapture
+
+        asynchronous_instruments_cumulative_with_gap_in_measurements_helper("counter");
+        asynchronous_instruments_cumulative_with_gap_in_measurements_helper("updown_counter");
+        asynchronous_instruments_cumulative_with_gap_in_measurements_helper("gauge");
+    }
+
+    fn asynchronous_instruments_cumulative_with_gap_in_measurements_helper(
+        instrument_name: &'static str,
+    ) {
+        let mut test_context = TestContext::new(Temporality::Cumulative);
+        let attributes = Arc::new([KeyValue::new("key1", "value1")]);
+
+        // Create instrument and emit measurements
+        match instrument_name {
+            "counter" => {
+                let has_run = AtomicBool::new(false);
+                let _observable_counter = test_context
+                    .meter()
+                    .u64_observable_counter("test_counter")
+                    .with_callback(move |observer| {
+                        if !has_run.load(Ordering::SeqCst) {
+                            observer.observe(5, &[]);
+                            observer.observe(10, &*attributes.clone());
+                            has_run.store(true, Ordering::SeqCst);
+                        }
+                    })
+                    .init();
+            }
+            "updown_counter" => {
+                let has_run = AtomicBool::new(false);
+                let _observable_up_down_counter = test_context
+                    .meter()
+                    .i64_observable_up_down_counter("test_updowncounter")
+                    .with_callback(move |observer| {
+                        if !has_run.load(Ordering::SeqCst) {
+                            observer.observe(15, &[]);
+                            observer.observe(20, &*attributes.clone());
+                            has_run.store(true, Ordering::SeqCst);
+                        }
+                    })
+                    .init();
+            }
+            "gauge" => {
+                let has_run = AtomicBool::new(false);
+                let _observable_gauge = test_context
+                    .meter()
+                    .u64_observable_gauge("test_gauge")
+                    .with_callback(move |observer| {
+                        if !has_run.load(Ordering::SeqCst) {
+                            observer.observe(25, &[]);
+                            observer.observe(30, &*attributes.clone());
+                            has_run.store(true, Ordering::SeqCst);
+                        }
+                    })
+                    .init();
+            }
+            _ => panic!("Incorrect instrument kind provided"),
+        };
+
+        test_context.flush_metrics();
+
+        // Test the first export
+        assert_correct_export(&mut test_context, instrument_name);
+
+        // Reset and export again without making any measurements
+        test_context.reset_metrics();
+
+        test_context.flush_metrics();
+
+        // Test that latest export has the same data as the previous one
+        assert_correct_export(&mut test_context, instrument_name);
+
+        fn assert_correct_export(test_context: &mut TestContext, instrument_name: &'static str) {
+            match instrument_name {
+                "counter" => {
+                    let counter_data =
+                        test_context.get_aggregation::<data::Sum<u64>>("test_counter", None);
+                    assert_eq!(counter_data.data_points.len(), 2);
+                    assert!(counter_data.is_monotonic);
+                    let zero_attribute_datapoint =
+                        find_datapoint_with_no_attributes(&counter_data.data_points)
+                            .expect("datapoint with no attributes expected");
+                    assert_eq!(zero_attribute_datapoint.value, 5);
+                    let data_point1 =
+                        find_datapoint_with_key_value(&counter_data.data_points, "key1", "value1")
+                            .expect("datapoint with key1=value1 expected");
+                    assert_eq!(data_point1.value, 10);
+                }
+                "updown_counter" => {
+                    let updown_counter_data =
+                        test_context.get_aggregation::<data::Sum<i64>>("test_updowncounter", None);
+                    assert_eq!(updown_counter_data.data_points.len(), 2);
+                    assert!(!updown_counter_data.is_monotonic);
+                    let zero_attribute_datapoint =
+                        find_datapoint_with_no_attributes(&updown_counter_data.data_points)
+                            .expect("datapoint with no attributes expected");
+                    assert_eq!(zero_attribute_datapoint.value, 15);
+                    let data_point1 = find_datapoint_with_key_value(
+                        &updown_counter_data.data_points,
+                        "key1",
+                        "value1",
+                    )
+                    .expect("datapoint with key1=value1 expected");
+                    assert_eq!(data_point1.value, 20);
+                }
+                "gauge" => {
+                    let gauge_data =
+                        test_context.get_aggregation::<data::Gauge<u64>>("test_gauge", None);
+                    assert_eq!(gauge_data.data_points.len(), 2);
+                    let zero_attribute_datapoint =
+                        find_datapoint_with_no_attributes(&gauge_data.data_points)
+                            .expect("datapoint with no attributes expected");
+                    assert_eq!(zero_attribute_datapoint.value, 25);
+                    let data_point1 =
+                        find_datapoint_with_key_value(&gauge_data.data_points, "key1", "value1")
+                            .expect("datapoint with key1=value1 expected");
+                    assert_eq!(data_point1.value, 30);
+                }
+                _ => panic!("Incorrect instrument kind provided"),
+            }
+        }
+    }
+
     fn counter_multithreaded_aggregation_helper(temporality: Temporality) {
         // Arrange
         let mut test_context = TestContext::new(temporality);