Implement MetricReaderStorage (#2456)

* Implement MetricReaderStorage

* Apply suggestions from code review

Co-authored-by: Diego Hurtado <[email protected]>

* fix tset

* syntax error

* move async instrument callback invocation into the metric reader storage

* Rename ViewStorage -> ViewInstrumentMatch

Tests still need to be fixed.

* Implement MetricReaderStorage

* Apply suggestions from code review

Co-authored-by: Diego Hurtado <[email protected]>

* fix tset

* syntax error

* move async instrument callback invocation into the metric reader storage

* Rename ViewStorage -> ViewInstrumentMatch

* fix lint

* fix lint

* Update opentelemetry-sdk/src/opentelemetry/sdk/_metrics/__init__.py

Co-authored-by: Diego Hurtado <[email protected]>

* refactor to have the measurement consumer handle async callbacks again

* remove print

* lint

Co-authored-by: Diego Hurtado <[email protected]>
Co-authored-by: Alex Boten <[email protected]>
Co-authored-by: Alex Boten <[email protected]>

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

@@ -159,7 +159,7 @@ def __init__(
         self._meter_lock = Lock()
         self._atexit_handler = None
         self._sdk_config = SdkConfiguration(
-            resource=resource, metric_readers=metric_readers
+            resource=resource, metric_readers=metric_readers, views=()
         )
         self._measurement_consumer = SynchronousMeasurementConsumer(
             sdk_config=self._sdk_config

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

@@ -95,7 +95,7 @@ def add(
             )
             return
         self._measurement_consumer.consume_measurement(
-            Measurement(amount, attributes)
+            Measurement(amount, self, attributes)
         )
 
 
@@ -104,7 +104,7 @@ def add(
         self, amount: Union[int, float], attributes: Dict[str, str] = None
     ):
         self._measurement_consumer.consume_measurement(
-            Measurement(amount, attributes)
+            Measurement(amount, self, attributes)
         )
 
 
@@ -127,7 +127,7 @@ def record(
             )
             return
         self._measurement_consumer.consume_measurement(
-            Measurement(amount, attributes)
+            Measurement(amount, self, attributes)
         )
 
 

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

@@ -15,10 +15,12 @@
 from dataclasses import dataclass
 from typing import Union
 
+from opentelemetry._metrics.instrument import Instrument
 from opentelemetry.util.types import Attributes
 
 
 @dataclass(frozen=True)
 class Measurement:
     value: Union[int, float]
+    instrument: Instrument
     attributes: Attributes = None

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

@@ -70,7 +70,8 @@ def collect(
         self, metric_reader: MetricReader, temporality: AggregationTemporality
     ) -> Iterable[Metric]:
         with self._lock:
+            metric_reader_storage = self._reader_storages[metric_reader]
             for async_instrument in self._async_instruments:
                 for measurement in async_instrument.callback():
-                    self.consume_measurement(measurement)
+                    metric_reader_storage.consume_measurement(measurement)
         return self._reader_storages[metric_reader].collect(temporality)

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

@@ -12,23 +12,112 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Iterable
+from threading import RLock
+from typing import Dict, Iterable, List
 
-from opentelemetry.sdk._metrics.aggregation import AggregationTemporality
+from opentelemetry._metrics.instrument import Counter, Histogram, Instrument
+from opentelemetry.sdk._metrics._view_instrument_match import (
+    _ViewInstrumentMatch,
+)
+from opentelemetry.sdk._metrics.aggregation import (
+    AggregationTemporality,
+    ExplicitBucketHistogramAggregation,
+    LastValueAggregation,
+    SumAggregation,
+)
 from opentelemetry.sdk._metrics.measurement import Measurement
 from opentelemetry.sdk._metrics.point import Metric
 from opentelemetry.sdk._metrics.sdk_configuration import SdkConfiguration
+from opentelemetry.sdk._metrics.view import View
 
 
-# TODO: #2378
 class MetricReaderStorage:
     """The SDK's storage for a given reader"""
 
     def __init__(self, sdk_config: SdkConfiguration) -> None:
-        pass
+        self._lock = RLock()
+        self._sdk_config = sdk_config
+        self._view_instrument_match: Dict[
+            Instrument, List[_ViewInstrumentMatch]
+        ] = {}
+
+    def _get_or_init_view_instrument_match(
+        self, instrument: Instrument
+    ) -> List["_ViewInstrumentMatch"]:
+        # Optimistically get the relevant views for the given instrument. Once set for a given
+        # instrument, the mapping will never change
+        if instrument in self._view_instrument_match:
+            return self._view_instrument_match[instrument]
+
+        with self._lock:
+            # double check if it was set before we held the lock
+            if instrument in self._view_instrument_match:
+                return self._view_instrument_match[instrument]
+
+            # not present, hold the lock and add a new mapping
+            matches = []
+            for view in self._sdk_config.views:
+                if view.match(instrument):
+                    # Note: if a view matches multiple instruments, this will create a separate
+                    # _ViewInstrumentMatch per instrument. If the user's View configuration includes a
+                    # name, this will cause multiple conflicting output streams.
+                    matches.append(
+                        _ViewInstrumentMatch(
+                            name=view.name or instrument.name,
+                            resource=self._sdk_config.resource,
+                            instrumentation_info=None,
+                            aggregation=view.aggregation,
+                            unit=instrument.unit,
+                            description=view.description,
+                        )
+                    )
+
+            # if no view targeted the instrument, use the default
+            if not matches:
+                # TODO: the logic to select aggregation could be moved
+                if isinstance(instrument, Counter):
+                    agg = SumAggregation(True, AggregationTemporality.DELTA)
+                elif isinstance(instrument, Histogram):
+                    agg = ExplicitBucketHistogramAggregation()
+                else:
+                    agg = LastValueAggregation()
+                matches.append(
+                    _ViewInstrumentMatch(
+                        resource=self._sdk_config.resource,
+                        instrumentation_info=None,
+                        aggregation=agg,
+                        unit=instrument.unit,
+                        description=instrument.description,
+                        name=instrument.name,
+                    )
+                )
+            self._view_instrument_match[instrument] = matches
+            return matches
 
     def consume_measurement(self, measurement: Measurement) -> None:
-        pass
+        for matches in self._get_or_init_view_instrument_match(
+            measurement.instrument
+        ):
+            matches.consume_measurement(measurement)
 
     def collect(self, temporality: AggregationTemporality) -> Iterable[Metric]:
-        pass
+        # use a list instead of yielding to prevent a slow reader from holding SDK locks
+        metrics: List[Metric] = []
+
+        # While holding the lock, new _ViewInstrumentMatch can't be added from another thread (so we are
+        # sure we collect all existing view). However, instruments can still send measurements
+        # that will make it into the individual aggregations; collection will acquire those
+        # locks iteratively to keep locking as fine-grained as possible. One side effect is
+        # that end times can be slightly skewed among the metric streams produced by the SDK,
+        # but we still align the output timestamps for a single instrument.
+        with self._lock:
+            for matches in self._view_instrument_match.values():
+                for match in matches:
+                    metrics.extend(match.collect(temporality))
+
+        return metrics
+
+
+def default_view(instrument: Instrument) -> View:
+    # TODO: #2247
+    return View()

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

@@ -2,12 +2,12 @@
 from typing import Sequence
 
 from opentelemetry.sdk._metrics.metric_reader import MetricReader
+from opentelemetry.sdk._metrics.view import View
 from opentelemetry.sdk.resources import Resource
 
 
 @dataclass
 class SdkConfiguration:
     resource: Resource
-    # TODO: once views are added
-    # views: Sequence[View]
     metric_readers: Sequence[MetricReader]
+    views: Sequence[View]

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

@@ -0,0 +1,20 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+# TODO: #2247
+# pylint: disable=no-self-use
+class View:
+    def match(self) -> bool:
+        return False