Handle min, max and sum in explicit bucket histogram aggregator

Author: ocelotl

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

@@ -14,10 +14,14 @@
 
 from abc import ABC, abstractmethod
 from collections import OrderedDict
+from logging import getLogger
 from math import inf
 
+from opentelemetry._metrics.instrument import _Monotonic
 from opentelemetry.util._time import _time_ns
 
+_logger = getLogger(__name__)
+
 
 class Aggregation(ABC):
     @property
@@ -37,8 +41,7 @@ class NoneAggregation(Aggregation):
     This aggregation drops all instrument measurements.
     """
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, instrument):
         self._value = None
 
     def aggregate(self, value):
@@ -50,8 +53,7 @@ class SumAggregation(Aggregation):
     This aggregation collects data for the SDK sum metric point.
     """
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, instrument):
         self._value = 0
 
     def aggregate(self, value):
@@ -64,8 +66,7 @@ class LastValueAggregation(Aggregation):
     This aggregation collects data for the SDK sum metric point.
     """
 
-    def __init__(self):
-        super().__init__()
+    def __init__(self, instrument):
         self._value = None
         self._timestamp = _time_ns()
 
@@ -82,14 +83,52 @@ class ExplicitBucketHistogramAggregation(Aggregation):
 
     def __init__(
         self,
+        instrument,
+        *args,
         boundaries=(0, 5, 10, 25, 50, 75, 100, 250, 500, 1000, inf),
         record_min_max=True,
     ):
         super().__init__()
         self._boundaries = boundaries
         self._value = OrderedDict([(key, 0) for key in boundaries])
+        self._min = inf
+        self._max = -inf
+        self._sum = 0
+        self._instrument = instrument
+        self._record_min_max = record_min_max
+
+    @property
+    def min(self):
+        if not self._record_min_max:
+            _logger.warning("Min is not being recorded")
+
+        return self._min
+
+    @property
+    def max(self):
+        if not self._record_min_max:
+            _logger.warning("Max is not being recorded")
+
+        return self._max
+
+    @property
+    def sum(self):
+        if isinstance(self._instrument, _Monotonic):
+            return self._sum
+
+        _logger.warning(
+            "Sum is not filled out when the associated "
+            "instrument is not monotonic"
+        )
 
     def aggregate(self, value):
+        if self._record_min_max:
+            self._min = min(self._min, value)
+            self._max = max(self._max, value)
+
+        if isinstance(self._instrument, _Monotonic):
+            self._sum += value
+
         for key in self._value.keys():
 
             if value < key:

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

@@ -46,11 +46,10 @@ def __init__(
         self._attributes_aggregations = {}
         self._aggregation = aggregation
         self._aggregation_config = aggregation_config
-        aggregation(**aggregation_config)
+        aggregation(self, **aggregation_config)
 
 
 class _Synchronous(_Instrument):
-
     def add(self, amount, attributes=None):
 
         if attributes is None:
@@ -59,8 +58,8 @@ def add(self, amount, attributes=None):
         attributes = frozenset(attributes.items())
         if attributes not in self._attributes_aggregations.keys():
 
-            self._attributes_aggregations[attributes] = (
-                self._aggregation(**self._aggregation_config)
+            self._attributes_aggregations[attributes] = self._aggregation(
+                self, **self._aggregation_config
             )
         self._attributes_aggregations[attributes].aggregate(amount)
 
@@ -79,7 +78,7 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )
 
 
@@ -97,7 +96,7 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )
 
 
@@ -116,7 +115,7 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )
 
 
@@ -135,7 +134,7 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )
 
 
@@ -153,7 +152,7 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )
 
 
@@ -172,5 +171,5 @@ def __init__(
             unit=unit,
             description=description,
             aggregation=aggregation,
-            aggregation_config=aggregation_config
+            aggregation_config=aggregation_config,
         )

opentelemetry-sdk/tests/metrics/test_aggregation.py

@@ -13,8 +13,10 @@
 # limitations under the License.
 
 
+from logging import WARNING
 from math import inf
 from unittest import TestCase
+from unittest.mock import Mock
 
 from opentelemetry.sdk._metrics.aggregation import (
     ExplicitBucketHistogramAggregation,
@@ -30,7 +32,7 @@ def test_aggregate(self):
         `NoneAggregation` drops all measurements.
         """
 
-        none_aggregation = NoneAggregation()
+        none_aggregation = NoneAggregation(Mock())
 
         none_aggregation.aggregate(1)
         none_aggregation.aggregate(2)
@@ -45,7 +47,7 @@ def test_aggregate(self):
         `SumAggregation` collects data for sum metric points
         """
 
-        sum_aggregation = SumAggregation()
+        sum_aggregation = SumAggregation(Mock())
 
         sum_aggregation.aggregate(1)
         sum_aggregation.aggregate(2)
@@ -61,7 +63,7 @@ def test_aggregate(self):
         temporality
         """
 
-        last_value_aggregation = LastValueAggregation()
+        last_value_aggregation = LastValueAggregation(Mock())
 
         last_value_aggregation.aggregate(1)
         self.assertEqual(last_value_aggregation.value, 1)
@@ -80,7 +82,7 @@ def test_aggregate(self):
         """
 
         explicit_bucket_histogram_aggregation = (
-            ExplicitBucketHistogramAggregation()
+            ExplicitBucketHistogramAggregation(Mock())
         )
 
         explicit_bucket_histogram_aggregation.aggregate(-1)
@@ -95,3 +97,38 @@ def test_aggregate(self):
         self.assertEqual(
             explicit_bucket_histogram_aggregation.value[inf], 9999
         )
+
+    def test_min_max(self):
+        """
+        `record_min_max` indicates the aggregator to record the minimum and
+        maximum value in the population
+        """
+
+        explicit_bucket_histogram_aggregation = (
+            ExplicitBucketHistogramAggregation(Mock())
+        )
+
+        explicit_bucket_histogram_aggregation.aggregate(-1)
+        explicit_bucket_histogram_aggregation.aggregate(2)
+        explicit_bucket_histogram_aggregation.aggregate(7)
+        explicit_bucket_histogram_aggregation.aggregate(8)
+        explicit_bucket_histogram_aggregation.aggregate(9999)
+
+        self.assertEqual(explicit_bucket_histogram_aggregation.min, -1)
+        self.assertEqual(explicit_bucket_histogram_aggregation.max, 9999)
+
+        explicit_bucket_histogram_aggregation = (
+            ExplicitBucketHistogramAggregation(Mock(), record_min_max=False)
+        )
+
+        explicit_bucket_histogram_aggregation.aggregate(-1)
+        explicit_bucket_histogram_aggregation.aggregate(2)
+        explicit_bucket_histogram_aggregation.aggregate(7)
+        explicit_bucket_histogram_aggregation.aggregate(8)
+        explicit_bucket_histogram_aggregation.aggregate(9999)
+
+        with self.assertLogs(level=WARNING):
+            self.assertEqual(explicit_bucket_histogram_aggregation.min, inf)
+
+        with self.assertLogs(level=WARNING):
+            self.assertEqual(explicit_bucket_histogram_aggregation.max, -inf)

opentelemetry-sdk/tests/metrics/test_instrument.py

@@ -15,12 +15,8 @@
 
 from unittest import TestCase
 
-from opentelemetry.sdk._metrics.instrument import (
-    _Synchronous
-)
-from opentelemetry.sdk._metrics.aggregation import (
-    SumAggregation
-)
+from opentelemetry.sdk._metrics.aggregation import SumAggregation
+from opentelemetry.sdk._metrics.instrument import _Synchronous
 
 
 class Test_Synchronous(TestCase):
@@ -38,11 +34,11 @@ def test_add(self):
             synchronous._attributes_aggregations[
                 frozenset({("name0", "value0")})
             ],
-            SumAggregation
+            SumAggregation,
         )
         self.assertIsInstance(
             synchronous._attributes_aggregations[
                 frozenset({("name1", "value1")})
             ],
-            SumAggregation
+            SumAggregation,
         )