Reset async instrumentation aggregations each collection interval (#2510)

Co-authored-by: Alex Boten <aboten@lightstep.com>

Author: aabmass

opentelemetry-sdk/src/opentelemetry/sdk/_metrics/aggregation.py

@@ -101,15 +101,18 @@ def collect(self) -> Optional[Sum]:
                 value=value,
             )
 
-        if self._value is None:
-            return None
+        with self._lock:
+            if self._value is None:
+                return None
+            value = self._value
+            self._value = None
 
         return Sum(
             aggregation_temporality=AggregationTemporality.CUMULATIVE,
             is_monotonic=self._instrument_is_monotonic,
             start_time_unix_nano=self._start_time_unix_nano,
             time_unix_nano=now,
-            value=self._value,
+            value=value,
         )
 
 
@@ -126,12 +129,15 @@ def collect(self) -> Optional[Gauge]:
         """
         Atomically return a point for the current value of the metric.
         """
-        if self._value is None:
-            return None
+        with self._lock:
+            if self._value is None:
+                return None
+            value = self._value
+            self._value = None
 
         return Gauge(
             time_unix_nano=_time_ns(),
-            value=self._value,
+            value=value,
         )
 
 

opentelemetry-sdk/tests/metrics/test_aggregation.py

@@ -133,19 +133,20 @@ def test_collect_cumulative(self):
         self.assertEqual(first_sum.value, 1)
         self.assertTrue(first_sum.is_monotonic)
 
+        # should have been reset after first collect
         sum_aggregation.aggregate(measurement(1))
         second_sum = sum_aggregation.collect()
 
-        self.assertEqual(second_sum.value, 2)
+        self.assertEqual(second_sum.value, 1)
         self.assertTrue(second_sum.is_monotonic)
 
         self.assertEqual(
             second_sum.start_time_unix_nano, first_sum.start_time_unix_nano
         )
 
-        self.assertIsNone(
-            _SumAggregation(True, AggregationTemporality.CUMULATIVE).collect()
-        )
+        # if no point seen for a whole interval, should return None
+        third_sum = sum_aggregation.collect()
+        self.assertIsNone(third_sum)
 
 
 class TestLastValueAggregation(TestCase):
@@ -194,6 +195,10 @@ def test_collect(self):
             second_gauge.time_unix_nano, first_gauge.time_unix_nano
         )
 
+        # if no observation seen for the interval, it should return None
+        third_gauge = last_value_aggregation.collect()
+        self.assertIsNone(third_gauge)
+
 
 class TestExplicitBucketHistogramAggregation(TestCase):
     def test_aggregate(self):