Remove instrument_class_temporality from collect args (#2674)

Fixes #2673

Author: ocelotl

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

@@ -43,6 +43,7 @@ def __init__(
         view: View,
         instrument: Instrument,
         sdk_config: SdkConfiguration,
+        instrument_class_temporality: Dict[type, AggregationTemporality],
         instrument_class_aggregation: Dict[type, Aggregation],
     ):
         self._view = view
@@ -51,6 +52,7 @@ def __init__(
         self._attributes_aggregation: Dict[frozenset, _Aggregation] = {}
         self._attributes_previous_point: Dict[frozenset, _PointVarT] = {}
         self._lock = Lock()
+        self._instrument_class_temporality = instrument_class_temporality
         self._instrument_class_aggregation = instrument_class_aggregation
         self._name = self._view._name or self._instrument.name
         self._description = (
@@ -122,9 +124,7 @@ def consume_measurement(self, measurement: Measurement) -> None:
 
         self._attributes_aggregation[attributes].aggregate(measurement)
 
-    def collect(
-        self, instrument_class_temporality: Dict[type, AggregationTemporality]
-    ) -> Iterable[Metric]:
+    def collect(self) -> Iterable[Metric]:
 
         with self._lock:
             for (
@@ -139,6 +139,7 @@ def collect(
                 current_point = aggregation.collect()
 
                 # pylint: disable=assignment-from-none
+
                 self._attributes_previous_point[
                     attributes
                 ] = _convert_aggregation_temporality(
@@ -161,7 +162,7 @@ def collect(
                         point=_convert_aggregation_temporality(
                             previous_point,
                             current_point,
-                            instrument_class_temporality[
+                            self._instrument_class_temporality[
                                 self._instrument.__class__
                             ],
                         ),

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

@@ -241,7 +241,7 @@ def collect(self, timeout_millis: float = 10_000) -> None:
             )
             return
         self._receive_metrics(
-            self._collect(self, self._instrument_class_temporality),
+            self._collect(self),
             timeout_millis=timeout_millis,
         )
 

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

@@ -16,14 +16,13 @@
 
 from abc import ABC, abstractmethod
 from threading import Lock
-from typing import Dict, Iterable, List, Mapping
+from typing import Iterable, List, Mapping
 
 # This kind of import is needed to avoid Sphinx errors.
 import opentelemetry.sdk._metrics
 import opentelemetry.sdk._metrics._internal.instrument
 import opentelemetry.sdk._metrics._internal.sdk_configuration
 from opentelemetry._metrics._internal.instrument import CallbackOptions
-from opentelemetry.sdk._metrics._internal.export import AggregationTemporality
 from opentelemetry.sdk._metrics._internal.measurement import Measurement
 from opentelemetry.sdk._metrics._internal.metric_reader_storage import (
     MetricReaderStorage,
@@ -49,7 +48,6 @@ def register_asynchronous_instrument(
     def collect(
         self,
         metric_reader: "opentelemetry.sdk._metrics.MetricReader",
-        instrument_type_temporality: Dict[type, AggregationTemporality],
     ) -> Iterable[Metric]:
         pass
 
@@ -66,7 +64,9 @@ def __init__(
             "opentelemetry.sdk._metrics.MetricReader", MetricReaderStorage
         ] = {
             reader: MetricReaderStorage(
-                sdk_config, reader._instrument_class_aggregation
+                sdk_config,
+                reader._instrument_class_temporality,
+                reader._instrument_class_aggregation,
             )
             for reader in sdk_config.metric_readers
         }
@@ -90,7 +90,6 @@ def register_asynchronous_instrument(
     def collect(
         self,
         metric_reader: "opentelemetry.sdk._metrics.MetricReader",
-        instrument_type_temporality: Dict[type, AggregationTemporality],
     ) -> Iterable[Metric]:
         with self._lock:
             metric_reader_storage = self._reader_storages[metric_reader]
@@ -99,6 +98,4 @@ def collect(
             for async_instrument in self._async_instruments:
                 for measurement in async_instrument.callback(callback_options):
                     metric_reader_storage.consume_measurement(measurement)
-        return self._reader_storages[metric_reader].collect(
-            instrument_type_temporality
-        )
+        return self._reader_storages[metric_reader].collect()

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

@@ -43,13 +43,15 @@ class MetricReaderStorage:
     def __init__(
         self,
         sdk_config: SdkConfiguration,
+        instrument_class_temporality: Dict[type, AggregationTemporality],
         instrument_class_aggregation: Dict[type, Aggregation],
     ) -> None:
         self._lock = RLock()
         self._sdk_config = sdk_config
         self._instrument_view_instrument_matches: Dict[
             Instrument, List[_ViewInstrumentMatch]
         ] = {}
+        self._instrument_class_temporality = instrument_class_temporality
         self._instrument_class_aggregation = instrument_class_aggregation
 
     def _get_or_init_view_instrument_match(
@@ -80,6 +82,9 @@ def _get_or_init_view_instrument_match(
                         view=_DEFAULT_VIEW,
                         instrument=instrument,
                         sdk_config=self._sdk_config,
+                        instrument_class_temporality=(
+                            self._instrument_class_temporality
+                        ),
                         instrument_class_aggregation=(
                             self._instrument_class_aggregation
                         ),
@@ -97,9 +102,7 @@ def consume_measurement(self, measurement: Measurement) -> None:
         ):
             view_instrument_match.consume_measurement(measurement)
 
-    def collect(
-        self, instrument_type_temporality: Dict[type, AggregationTemporality]
-    ) -> Iterable[Metric]:
+    def collect(self) -> Iterable[Metric]:
         # Use a list instead of yielding to prevent a slow reader from holding
         # SDK locks
         metrics: List[Metric] = []
@@ -117,11 +120,7 @@ def collect(
                 view_instrument_matches
             ) in self._instrument_view_instrument_matches.values():
                 for view_instrument_match in view_instrument_matches:
-                    metrics.extend(
-                        view_instrument_match.collect(
-                            instrument_type_temporality
-                        )
-                    )
+                    metrics.extend(view_instrument_match.collect())
 
         return metrics
 
@@ -142,6 +141,9 @@ def _handle_view_instrument_match(
                 view=view,
                 instrument=instrument,
                 sdk_config=self._sdk_config,
+                instrument_class_temporality=(
+                    self._instrument_class_temporality
+                ),
                 instrument_class_aggregation=(
                     self._instrument_class_aggregation
                 ),

opentelemetry-sdk/tests/metrics/test_measurement_consumer.py

@@ -22,7 +22,6 @@
 from opentelemetry.sdk._metrics._internal.sdk_configuration import (
     SdkConfiguration,
 )
-from opentelemetry.sdk._metrics.export import AggregationTemporality
 
 
 @patch(
@@ -85,10 +84,8 @@ def test_collect_passed_to_reader_stage(self, MockMetricReaderStorage):
         )
         for r_mock, rs_mock in zip(reader_mocks, reader_storage_mocks):
             rs_mock.collect.assert_not_called()
-            consumer.collect(r_mock, AggregationTemporality.CUMULATIVE)
-            rs_mock.collect.assert_called_once_with(
-                AggregationTemporality.CUMULATIVE
-            )
+            consumer.collect(r_mock)
+            rs_mock.collect.assert_called_once_with()
 
     def test_collect_calls_async_instruments(self, MockMetricReaderStorage):
         """Its collect() method should invoke async instruments and pass measurements to the
@@ -108,7 +105,7 @@ def test_collect_calls_async_instruments(self, MockMetricReaderStorage):
             i_mock.callback.return_value = [Mock()]
             consumer.register_asynchronous_instrument(i_mock)
 
-        consumer.collect(reader_mock, AggregationTemporality.CUMULATIVE)
+        consumer.collect(reader_mock)
 
         # it should call async instruments
         for i_mock in async_instrument_mocks: