Implement metric reader default aggregation controls

Fixes #2635

Author: ocelotl

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

@@ -18,7 +18,9 @@
 from typing import TYPE_CHECKING, Dict, Iterable
 
 from opentelemetry.sdk._metrics.aggregation import (
+    DefaultAggregation,
     _Aggregation,
+    _AggregationFactory,
     _convert_aggregation_temporality,
     _PointVarT,
 )
@@ -39,13 +41,15 @@ def __init__(
         view: View,
         instrument: "_Instrument",
         sdk_config: SdkConfiguration,
+        instrument_class_aggregation: Dict[type, _AggregationFactory],
     ):
         self._view = view
         self._instrument = instrument
         self._sdk_config = sdk_config
         self._attributes_aggregation: Dict[frozenset, _Aggregation] = {}
         self._attributes_previous_point: Dict[frozenset, _PointVarT] = {}
         self._lock = Lock()
+        self._instrument_class_aggregation = instrument_class_aggregation
 
     # pylint: disable=protected-access
     def consume_measurement(self, measurement: Measurement) -> None:
@@ -67,11 +71,19 @@ def consume_measurement(self, measurement: Measurement) -> None:
         if attributes not in self._attributes_aggregation:
             with self._lock:
                 if attributes not in self._attributes_aggregation:
-                    self._attributes_aggregation[
-                        attributes
-                    ] = self._view._aggregation._create_aggregation(
-                        self._instrument
-                    )
+                    if not isinstance(
+                        self._view._aggregation, DefaultAggregation
+                    ):
+                        aggregation = (
+                            self._view._aggregation._create_aggregation(
+                                self._instrument
+                            )
+                        )
+                    else:
+                        aggregation = self._instrument_class_aggregation[
+                            self._instrument.__class__
+                        ]._create_aggregation(self._instrument)
+                    self._attributes_aggregation[attributes] = aggregation
 
         self._attributes_aggregation[attributes].aggregate(measurement)
 

opentelemetry-sdk/src/opentelemetry/sdk/_metrics/export/__init__.py

@@ -19,14 +19,15 @@
 from os import environ, linesep
 from sys import stdout
 from threading import Event, RLock, Thread
-from typing import IO, Callable, Iterable, List, Optional, Sequence
+from typing import IO, Callable, Dict, Iterable, List, Optional, Sequence
 
 from opentelemetry.context import (
     _SUPPRESS_INSTRUMENTATION_KEY,
     attach,
     detach,
     set_value,
 )
+from opentelemetry.sdk._metrics.aggregation import _AggregationFactory
 from opentelemetry.sdk._metrics.metric_reader import MetricReader
 from opentelemetry.sdk._metrics.point import AggregationTemporality, Metric
 from opentelemetry.util._once import Once
@@ -104,9 +105,13 @@ class InMemoryMetricReader(MetricReader):
 
     def __init__(
         self,
+        preferred_aggregation: Dict[type, _AggregationFactory] = None,
         preferred_temporality: AggregationTemporality = AggregationTemporality.CUMULATIVE,
     ) -> None:
-        super().__init__(preferred_temporality=preferred_temporality)
+        super().__init__(
+            preferred_temporality=preferred_temporality,
+            preferred_aggregation=preferred_aggregation,
+        )
         self._lock = RLock()
         self._metrics: List[Metric] = []
 
@@ -135,10 +140,14 @@ class PeriodicExportingMetricReader(MetricReader):
     def __init__(
         self,
         exporter: MetricExporter,
+        preferred_aggregation: Dict[type, _AggregationFactory] = None,
         export_interval_millis: Optional[float] = None,
         export_timeout_millis: Optional[float] = None,
     ) -> None:
-        super().__init__(preferred_temporality=exporter.preferred_temporality)
+        super().__init__(
+            preferred_temporality=exporter.preferred_temporality,
+            preferred_aggregation=preferred_aggregation,
+        )
         self._exporter = exporter
         if export_interval_millis is None:
             try:

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

@@ -14,8 +14,8 @@
 
 # pylint: disable=too-many-ancestors
 
-import logging
-from typing import Dict, Generator, Iterable, Optional, Union
+from logging import getLogger
+from typing import TYPE_CHECKING, Dict, Generator, Iterable, Optional, Union
 
 from opentelemetry._metrics.instrument import CallbackT
 from opentelemetry._metrics.instrument import Counter as APICounter
@@ -31,18 +31,23 @@
 )
 from opentelemetry._metrics.instrument import UpDownCounter as APIUpDownCounter
 from opentelemetry.sdk._metrics.measurement import Measurement
-from opentelemetry.sdk._metrics.measurement_consumer import MeasurementConsumer
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
 
-_logger = logging.getLogger(__name__)
+if TYPE_CHECKING:
+    from opentelemetry.sdk._metrics.measurement_consumer import (
+        MeasurementConsumer,
+    )
+
+
+_logger = getLogger(__name__)
 
 
 class _Synchronous:
     def __init__(
         self,
         name: str,
         instrumentation_scope: InstrumentationScope,
-        measurement_consumer: MeasurementConsumer,
+        measurement_consumer: "MeasurementConsumer",
         unit: str = "",
         description: str = "",
     ):
@@ -59,7 +64,7 @@ def __init__(
         self,
         name: str,
         instrumentation_scope: InstrumentationScope,
-        measurement_consumer: MeasurementConsumer,
+        measurement_consumer: "MeasurementConsumer",
         callbacks: Optional[Iterable[CallbackT]] = None,
         unit: str = "",
         description: str = "",

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

@@ -51,7 +51,9 @@ def __init__(self, sdk_config: SdkConfiguration) -> None:
         self._sdk_config = sdk_config
         # should never be mutated
         self._reader_storages: Mapping[MetricReader, MetricReaderStorage] = {
-            reader: MetricReaderStorage(sdk_config)
+            reader: MetricReaderStorage(
+                sdk_config, reader._instrument_class_aggregation
+            )
             for reader in sdk_config.metric_readers
         }
         self._async_instruments: List["_Asynchronous"] = []

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

@@ -14,29 +14,68 @@
 
 import logging
 from abc import ABC, abstractmethod
-from typing import Callable, Iterable
+from typing import Callable, Dict, Iterable
 
 from typing_extensions import final
 
+from opentelemetry.sdk._metrics.aggregation import (
+    DefaultAggregation,
+    _AggregationFactory,
+)
+from opentelemetry.sdk._metrics.instrument import (
+    Counter,
+    Histogram,
+    ObservableCounter,
+    ObservableGauge,
+    ObservableUpDownCounter,
+    UpDownCounter,
+)
 from opentelemetry.sdk._metrics.point import AggregationTemporality, Metric
 
 _logger = logging.getLogger(__name__)
 
 
 class MetricReader(ABC):
     """
+    Base class for all metric readers
+
+    Args:
+        preferred_aggregation: A mapping between instrument classes and
+            aggregation instances. By default maps all instrument classes to an
+            instance of `DefaultAggregation`. This mapping will be used to
+            define the default aggregation of every instrument class. If the
+            user wants to make a change in the default aggregation of an
+            instrument class, it is enough to pass here a dictionary whose keys
+            are the instrument classes and the values are the corresponding
+            desired aggregation for the instrument classes that the user wants
+            to change, not necessarily all of them. The classes not included in
+            the passed dictionary will retain their association to their
+            default aggregations.
+
     .. document protected _receive_metrics which is a intended to be overriden by subclass
     .. automethod:: _receive_metrics
     """
 
     def __init__(
         self,
         preferred_temporality: AggregationTemporality = AggregationTemporality.CUMULATIVE,
+        preferred_aggregation: Dict[type, _AggregationFactory] = None,
     ) -> None:
         self._collect: Callable[
             ["MetricReader", AggregationTemporality], Iterable[Metric]
         ] = None
+
         self._preferred_temporality = preferred_temporality
+        self._instrument_class_aggregation = {
+            Counter: DefaultAggregation(),
+            UpDownCounter: DefaultAggregation(),
+            Histogram: DefaultAggregation(),
+            ObservableCounter: DefaultAggregation(),
+            ObservableUpDownCounter: DefaultAggregation(),
+            ObservableGauge: DefaultAggregation(),
+        }
+
+        self._instrument_class_aggregation.update(preferred_aggregation or {})
 
     @final
     def collect(self) -> None:

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

@@ -19,7 +19,10 @@
 from opentelemetry.sdk._metrics._view_instrument_match import (
     _ViewInstrumentMatch,
 )
-from opentelemetry.sdk._metrics.aggregation import AggregationTemporality
+from opentelemetry.sdk._metrics.aggregation import (
+    AggregationTemporality,
+    _AggregationFactory,
+)
 from opentelemetry.sdk._metrics.measurement import Measurement
 from opentelemetry.sdk._metrics.point import Metric
 from opentelemetry.sdk._metrics.sdk_configuration import SdkConfiguration
@@ -31,12 +34,17 @@
 class MetricReaderStorage:
     """The SDK's storage for a given reader"""
 
-    def __init__(self, sdk_config: SdkConfiguration) -> None:
+    def __init__(
+        self,
+        sdk_config: SdkConfiguration,
+        instrument_class_aggregation: Dict[type, _AggregationFactory],
+    ) -> None:
         self._lock = RLock()
         self._sdk_config = sdk_config
         self._view_instrument_match: Dict[
             Instrument, List[_ViewInstrumentMatch]
         ] = {}
+        self._instrument_class_aggregation = instrument_class_aggregation
 
     def _get_or_init_view_instrument_match(
         self, instrument: Instrument
@@ -62,6 +70,9 @@ def _get_or_init_view_instrument_match(
                             view=view,
                             instrument=instrument,
                             sdk_config=self._sdk_config,
+                            instrument_class_aggregation=(
+                                self._instrument_class_aggregation
+                            ),
                         )
                     )
 
@@ -72,6 +83,9 @@ def _get_or_init_view_instrument_match(
                         view=_DEFAULT_VIEW,
                         instrument=instrument,
                         sdk_config=self._sdk_config,
+                        instrument_class_aggregation=(
+                            self._instrument_class_aggregation
+                        ),
                     )
                 )
             self._view_instrument_match[instrument] = view_instrument_matches

opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py

@@ -12,9 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
-from opentelemetry.sdk._metrics.aggregation import DropAggregation
+from opentelemetry.sdk._metrics.aggregation import (
+    DefaultAggregation,
+    DropAggregation,
+)
 from opentelemetry.sdk._metrics.instrument import Counter
 from opentelemetry.sdk._metrics.measurement import Measurement
 from opentelemetry.sdk._metrics.metric_reader_storage import (
@@ -56,7 +59,8 @@ def test_creates_view_instrument_matches(
                 resource=Mock(),
                 metric_readers=(),
                 views=(view1, view2),
-            )
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
         )
 
         # instrument1 matches view1 and view2, so should create two ViewInstrumentMatch objects
@@ -100,7 +104,8 @@ def test_forwards_calls_to_view_instrument_match(
                 resource=Mock(),
                 metric_readers=(),
                 views=(view1, view2),
-            )
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
         )
 
         # Measurements from an instrument should be passed on to each ViewInstrumentMatch objects
@@ -147,7 +152,8 @@ def test_race_concurrent_measurements(self, MockViewInstrumentMatch: Mock):
                 resource=Mock(),
                 metric_readers=(),
                 views=(view1,),
-            )
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
         )
 
         def send_measurement():
@@ -172,7 +178,8 @@ def test_default_view_enabled(self, MockViewInstrumentMatch: Mock):
                 resource=Mock(),
                 metric_readers=(),
                 views=(),
-            )
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
         )
 
         storage.consume_measurement(Measurement(1, instrument1))
@@ -200,7 +207,8 @@ def test_drop_aggregation(self):
                         instrument_name="name", aggregation=DropAggregation()
                     ),
                 ),
-            )
+            ),
+            MagicMock(**{"__getitem__.return_value": DefaultAggregation()}),
         )
         metric_reader_storage.consume_measurement(Measurement(1, counter))
 

opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py

@@ -84,7 +84,9 @@ def _create_periodic_reader(
         self, metrics, exporter, collect_wait=0, interval=60000
     ):
 
-        pmr = PeriodicExportingMetricReader(exporter, interval)
+        pmr = PeriodicExportingMetricReader(
+            exporter, export_interval_millis=interval
+        )
 
         def _collect(reader, temp):
             time.sleep(collect_wait)
@@ -95,7 +97,7 @@ def _collect(reader, temp):
 
     def test_ticker_called(self):
         collect_mock = Mock()
-        pmr = PeriodicExportingMetricReader(Mock(), 1)
+        pmr = PeriodicExportingMetricReader(Mock(), export_interval_millis=1)
         pmr._set_collect_callback(collect_mock)
         time.sleep(0.1)
         self.assertTrue(collect_mock.assert_called_once)