Add MetricReader

Fixes #2295

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

@@ -0,0 +1,37 @@
+# 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.
+
+from typing import Callable, Dict, List
+
+from opentelemetry.sdk._metrics.aggregation import Aggregation
+from opentelemetry.sdk._metrics.measurement import Measurement
+from opentelemetry.sdk.resources import Resource
+
+
+class _ViewInstrumentMatch:
+    def __init__(
+        self,
+        name: str,
+        unit: str,
+        description: str,
+        attribute_keys: Dict[str, str],
+        extra_dimensions: List[str],
+        aggregation: Aggregation,
+        exemplar_reservoir: Callable,
+        resource: Resource,
+    ):
+        pass
+
+    def _process(self, measurement: Measurement) -> None:
+        pass

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

@@ -0,0 +1,26 @@
+# 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.
+
+# FIXME this is being copied directly from
+# opentelemetry.proto.metrics.v1.metrics_pb2. The only reason for doing so is
+# to avoid havinv protobuf as a indirect dependency in the SDK. This
+# duplication of code is not ideal.
+
+AGGREGATION_TEMPORALITY_UNSPECIFIED = 0
+AGGREGATION_TEMPORALITY_DELTA = 1
+AGGREGATION_TEMPORALITY_CUMULATIVE = 2
+
+
+class Metric:
+    pass

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

@@ -12,6 +12,120 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from logging import getLogger
+from threading import RLock
+from typing import Dict, Iterable, List
+
+from opentelemetry._metrics import MeterProvider
+from opentelemetry._metrics.instrument import Instrument, Synchronous
+from opentelemetry.sdk._metrics._view_instrument_match import (
+    _ViewInstrumentMatch,
+)
+from opentelemetry.sdk._metrics.export import (
+    AGGREGATION_TEMPORALITY_CUMULATIVE,
+    Metric,
+)
+from opentelemetry.sdk._metrics.measurement import Measurement
+
+_logger = getLogger(__name__)
+
 
 class MetricReader:
-    pass
+    def __init__(
+        self, aggregation_temporality: int = AGGREGATION_TEMPORALITY_CUMULATIVE
+    ) -> None:
+
+        self._meter_provider = None
+        self._rlock = RLock()
+        self._aggregation_temporality = aggregation_temporality
+        self._instrument_view_instrument_matches: Dict[
+            Instrument, List[_ViewInstrumentMatch]
+        ] = {}
+
+    def _register_meter_provider(self, meter_provider: MeterProvider):
+
+        self._meter_provider = meter_provider
+
+    def _process(
+        self, instrument: Instrument, measurement: Measurement
+    ) -> List[_ViewInstrumentMatch]:
+
+        # pylint: disable=consider-iterating-dictionary
+        if instrument in self._instrument_view_instrument_matches.keys():
+            view_instrument_matches = self._instrument_view_instrument_matches[
+                instrument
+            ]
+
+        else:
+
+            with self._rlock:
+
+                if instrument in (
+                    # pylint: disable=consider-iterating-dictionary
+                    self._instrument_view_instrument_matches.keys()
+                ):
+                    view_instrument_matches = (
+                        self._instrument_view_instrument_matches[instrument]
+                    )
+
+                else:
+
+                    view_instrument_matches = []
+
+                    # pylint: disable=protected-access
+                    for view in self._meter_provider._views:
+
+                        if view._matches(instrument):
+                            view_instrument_matches.append(
+                                _ViewInstrumentMatch(
+                                    view._name or instrument._name,
+                                    view._unit or instrument._unit,
+                                    (
+                                        view._description
+                                        or instrument._description
+                                    ),
+                                    view._attribute_keys or {},
+                                    (
+                                        view._extra_dimensions
+                                        if isinstance(instrument, Synchronous)
+                                        else None
+                                    ),
+                                    (
+                                        view._aggregation
+                                        or instrument._default_aggregation
+                                    ),
+                                    view._exemplar_reservoir,
+                                    self._meter_provider._resource,
+                                )
+                            )
+
+        self._instrument_view_instrument_matches[
+            instrument
+        ] = view_instrument_matches
+
+        for view_instrument_match in view_instrument_matches:
+
+            # pylint: disable=protected-access
+            view_instrument_match._process(measurement)
+
+    def collect(self) -> Iterable[Metric]:
+        if self._meter_provider is None:
+            _logger.warning(
+                "Can't call collect on a MetricReader "
+                "until it is registered on a MeterProvider"
+            )
+
+            return []
+
+        # FIXME make async instruments report their last measurement
+
+        metrics: List[Metric] = []
+        # pylint: disable=pointless-statement
+        metrics
+
+        # FIXME collect from ViewStorages in original design
+
+        return []
+
+    def shutdown(self) -> None:
+        pass

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

@@ -12,7 +12,28 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Callable, Dict, List
+
+from opentelemetry._metrics.instrument import Instrument
+from opentelemetry.sdk._metrics.aggregation import Aggregation
+
 
 class View:
-    def __init__(self, instrument_name=None):
+    def __init__(
+        self,
+        instrument_type: Instrument = None,
+        instrument_name: str = None,
+        meter_name: str = None,
+        meter_version: str = None,
+        meter_schema_url: str = None,
+        name: str = None,
+        description: str = None,
+        attribute_keys: Dict[str, str] = None,
+        extra_dimensions: List[str] = None,
+        aggregation: Aggregation = None,
+        exemplar_reservoir: Callable = None,
+    ):
+        pass
+
+    def _matches(self, instrument: Instrument) -> bool:
         pass