Fix ViewInstrumentMatch to create new aggregation for each attribute set (#2503)

Author: aabmass

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

@@ -15,50 +15,45 @@
 
 from logging import getLogger
 from threading import Lock
-from typing import Iterable, Set
+from typing import TYPE_CHECKING, Iterable
 
 from opentelemetry.sdk._metrics.aggregation import (
-    _Aggregation,
     _convert_aggregation_temporality,
 )
 from opentelemetry.sdk._metrics.measurement import Measurement
 from opentelemetry.sdk._metrics.point import AggregationTemporality, Metric
-from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.util.instrumentation import InstrumentationInfo
+from opentelemetry.sdk._metrics.sdk_configuration import SdkConfiguration
+from opentelemetry.sdk._metrics.view import View
+
+if TYPE_CHECKING:
+    from opentelemetry.sdk._metrics.instrument import _Instrument
 
 _logger = getLogger(__name__)
 
 
 class _ViewInstrumentMatch:
     def __init__(
         self,
-        name: str,
-        unit: str,
-        description: str,
-        aggregation: _Aggregation,
-        instrumentation_info: InstrumentationInfo,
-        resource: Resource,
-        attribute_keys: Set[str],
+        view: View,
+        instrument: "_Instrument",
+        sdk_config: SdkConfiguration,
     ):
-        self._name = name
-        self._unit = unit
-        self._description = description
-        self._aggregation = aggregation
-        self._instrumentation_info = instrumentation_info
-        self._resource = resource
-        self._attribute_keys = attribute_keys
+        self._view = view
+        self._instrument = instrument
+        self._sdk_config = sdk_config
         self._attributes_aggregation = {}
         self._attributes_previous_point = {}
         self._lock = Lock()
 
+    # pylint: disable=protected-access
     def consume_measurement(self, measurement: Measurement) -> None:
 
-        if self._attribute_keys:
+        if self._view._attribute_keys is not None:
 
             attributes = {}
 
-            for key, value in measurement.attributes.items():
-                if key in self._attribute_keys:
+            for key, value in (measurement.attributes or {}).items():
+                if key in self._view._attribute_keys:
                     attributes[key] = value
         elif measurement.attributes is not None:
             attributes = measurement.attributes
@@ -67,9 +62,18 @@ def consume_measurement(self, measurement: Measurement) -> None:
 
         attributes = frozenset(attributes.items())
 
-        if attributes not in self._attributes_aggregation.keys():
+        if attributes not in self._attributes_aggregation:
             with self._lock:
-                self._attributes_aggregation[attributes] = self._aggregation
+                if attributes not in self._attributes_aggregation:
+                    if self._view._aggregation:
+                        aggregation = (
+                            self._view._aggregation._create_aggregation(
+                                self._instrument
+                            )
+                        )
+                    else:
+                        aggregation = self._instrument._default_aggregation
+                    self._attributes_aggregation[attributes] = aggregation
 
         self._attributes_aggregation[attributes].aggregate(measurement)
 
@@ -100,11 +104,14 @@ def collect(self, temporality: int) -> Iterable[Metric]:
 
                     yield Metric(
                         attributes=dict(attributes),
-                        description=self._description,
-                        instrumentation_info=self._instrumentation_info,
-                        name=self._name,
-                        resource=self._resource,
-                        unit=self._unit,
+                        description=(
+                            self._view._description
+                            or self._instrument.description
+                        ),
+                        instrumentation_info=self._instrument.instrumentation_info,
+                        name=self._view._name or self._instrument.name,
+                        resource=self._sdk_config.resource,
+                        unit=self._instrument.unit,
                         point=_convert_aggregation_temporality(
                             previous_point,
                             current_point,

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

@@ -13,14 +13,16 @@
 # limitations under the License.
 
 from dataclasses import dataclass
-from typing import Union
+from typing import TYPE_CHECKING, Union
 
-from opentelemetry._metrics.instrument import Instrument
 from opentelemetry.util.types import Attributes
 
+if TYPE_CHECKING:
+    from opentelemetry.sdk._metrics.instrument import _Instrument
+
 
 @dataclass(frozen=True)
 class Measurement:
     value: Union[int, float]
-    instrument: Instrument
+    instrument: "_Instrument"
     attributes: Attributes = None

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

@@ -25,6 +25,8 @@
 from opentelemetry.sdk._metrics.sdk_configuration import SdkConfiguration
 from opentelemetry.sdk._metrics.view import View
 
+_DEFAULT_VIEW = View(instrument_name="")
+
 
 class MetricReaderStorage:
     """The SDK's storage for a given reader"""
@@ -55,27 +57,11 @@ def _get_or_init_view_instrument_match(
             for view in self._sdk_config.views:
                 # pylint: disable=protected-access
                 if view._match(instrument):
-
-                    if view._aggregation is not None:
-                        aggregation = view._aggregation._create_aggregation(
-                            instrument
-                        )
-                    else:
-                        aggregation = instrument._default_aggregation
-
                     view_instrument_matches.append(
                         _ViewInstrumentMatch(
-                            name=view._name or instrument.name,
-                            unit=instrument.unit,
-                            description=(
-                                view._description or instrument.description
-                            ),
-                            aggregation=aggregation,
-                            instrumentation_info=(
-                                instrument.instrumentation_info
-                            ),
-                            resource=self._sdk_config.resource,
-                            attribute_keys=view._attribute_keys,
+                            view=view,
+                            instrument=instrument,
+                            sdk_config=self._sdk_config,
                         )
                     )
 
@@ -86,14 +72,9 @@ def _get_or_init_view_instrument_match(
             ):
                 view_instrument_matches.append(
                     _ViewInstrumentMatch(
-                        name=instrument.name,
-                        unit=instrument.unit,
-                        description=instrument.description,
-                        # pylint: disable=protected-access
-                        aggregation=instrument._default_aggregation,
-                        instrumentation_info=instrument.instrumentation_info,
-                        resource=self._sdk_config.resource,
-                        attribute_keys=set(),
+                        view=_DEFAULT_VIEW,
+                        instrument=instrument,
+                        sdk_config=self._sdk_config,
                     )
                 )
             self._view_instrument_match[instrument] = view_instrument_matches

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

@@ -127,10 +127,6 @@ class the instrument must be to match the view.
 
         self._description = description
         self._attribute_keys = attribute_keys
-
-        if self._attribute_keys is None:
-            self._attribute_keys = set()
-
         self._aggregation = aggregation
 
     # pylint: disable=too-many-return-statements

opentelemetry-sdk/tests/metrics/test_metric_reader_storage.py

@@ -14,15 +14,12 @@
 
 from unittest.mock import Mock, patch
 
-from opentelemetry.sdk._metrics.aggregation import SumAggregation
-from opentelemetry.sdk._metrics.instrument import Counter
 from opentelemetry.sdk._metrics.measurement import Measurement
 from opentelemetry.sdk._metrics.metric_reader_storage import (
     MetricReaderStorage,
 )
 from opentelemetry.sdk._metrics.point import AggregationTemporality
 from opentelemetry.sdk._metrics.sdk_configuration import SdkConfiguration
-from opentelemetry.sdk._metrics.view import View
 from opentelemetry.test.concurrency_test import ConcurrencyTestBase, MockFunc
 
 
@@ -162,38 +159,6 @@ def send_measurement():
         # _ViewInstrumentMatch constructor should have only been called once
         self.assertEqual(mock_view_instrument_match_ctor.call_count, 1)
 
-    def test_aggregations(self):
-
-        view = View(instrument_name="counter*", aggregation=SumAggregation())
-
-        metric_reader_storage = MetricReaderStorage(
-            SdkConfiguration(
-                resource=Mock(),
-                metric_readers=(),
-                views=(view,),
-                enable_default_view=True,
-            )
-        )
-
-        counter_0 = Counter("counter_0", Mock(), Mock())
-        counter_1 = Counter("counter_1", Mock(), Mock())
-
-        metric_reader_storage.consume_measurement(Measurement(1, counter_0))
-        metric_reader_storage.consume_measurement(Measurement(2, counter_1))
-
-        self.assertEqual(
-            1,
-            metric_reader_storage._view_instrument_match[counter_0][
-                0
-            ]._aggregation._value,
-        )
-        self.assertEqual(
-            2,
-            metric_reader_storage._view_instrument_match[counter_1][
-                0
-            ]._aggregation._value,
-        )
-
     @patch(
         "opentelemetry.sdk._metrics.metric_reader_storage._ViewInstrumentMatch"
     )