Respect OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE only for otlp exporter (#2843)

Author: lzchen

CHANGELOG.md

@@ -19,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#2754](https://github.com/open-telemetry/opentelemetry-python/pull/2754))
 - Fix --insecure of CLI argument
   ([#2696](https://github.com/open-telemetry/opentelemetry-python/pull/2696))
+- Add temporality and aggregation configuration for metrics exporters,
+  use `OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE` only for OTLP metrics exporter
+  ([#2843](https://github.com/open-telemetry/opentelemetry-python/pull/2843))
 -  Instrument instances are always created through a Meter
   ([#2844](https://github.com/open-telemetry/opentelemetry-python/pull/2844))
 

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/metric_exporter/__init__.py

@@ -13,8 +13,9 @@
 
 from logging import getLogger
 from os import environ
-from typing import Optional, Sequence
+from typing import Dict, Optional, Sequence
 from grpc import ChannelCredentials, Compression
+from opentelemetry.sdk.metrics._internal.aggregation import Aggregation
 from opentelemetry.exporter.otlp.proto.grpc.exporter import (
     OTLPExporterMixin,
     get_resource_data,
@@ -29,18 +30,25 @@
 from opentelemetry.proto.metrics.v1 import metrics_pb2 as pb2
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_METRICS_INSECURE,
+    OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
 )
-from opentelemetry.sdk.metrics.export import (
-    Gauge,
+from opentelemetry.sdk.metrics import (
+    Counter,
     Histogram,
-    Metric,
-    Sum,
+    ObservableCounter,
+    ObservableGauge,
+    ObservableUpDownCounter,
+    UpDownCounter,
 )
-
 from opentelemetry.sdk.metrics.export import (
+    AggregationTemporality,
+    Gauge,
+    Histogram as HistogramType,
+    Metric,
     MetricExporter,
     MetricExportResult,
     MetricsData,
+    Sum,
 )
 
 _logger = getLogger(__name__)
@@ -61,22 +69,57 @@ def __init__(
         headers: Optional[Sequence] = None,
         timeout: Optional[int] = None,
         compression: Optional[Compression] = None,
+        preferred_temporality: Dict[type, AggregationTemporality] = None,
+        preferred_aggregation: Dict[type, Aggregation] = None,
     ):
 
         if insecure is None:
             insecure = environ.get(OTEL_EXPORTER_OTLP_METRICS_INSECURE)
             if insecure is not None:
                 insecure = insecure.lower() == "true"
 
-        super().__init__(
-            **{
-                "endpoint": endpoint,
-                "insecure": insecure,
-                "credentials": credentials,
-                "headers": headers,
-                "timeout": timeout,
-                "compression": compression,
+        instrument_class_temporality = {}
+        if (
+            environ.get(
+                OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
+                "CUMULATIVE",
+            )
+            .upper()
+            .strip()
+            == "DELTA"
+        ):
+            instrument_class_temporality = {
+                Counter: AggregationTemporality.DELTA,
+                UpDownCounter: AggregationTemporality.CUMULATIVE,
+                Histogram: AggregationTemporality.DELTA,
+                ObservableCounter: AggregationTemporality.DELTA,
+                ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
+                ObservableGauge: AggregationTemporality.CUMULATIVE,
             }
+        else:
+            instrument_class_temporality = {
+                Counter: AggregationTemporality.CUMULATIVE,
+                UpDownCounter: AggregationTemporality.CUMULATIVE,
+                Histogram: AggregationTemporality.CUMULATIVE,
+                ObservableCounter: AggregationTemporality.CUMULATIVE,
+                ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
+                ObservableGauge: AggregationTemporality.CUMULATIVE,
+            }
+        instrument_class_temporality.update(preferred_temporality or {})
+
+        MetricExporter.__init__(
+            self,
+            preferred_temporality=instrument_class_temporality,
+            preferred_aggregation=preferred_aggregation,
+        )
+        OTLPExporterMixin.__init__(
+            self,
+            endpoint=endpoint,
+            insecure=insecure,
+            credentials=credentials,
+            headers=headers,
+            timeout=timeout,
+            compression=compression,
         )
 
     def _translate_data(
@@ -132,7 +175,7 @@ def _translate_data(
                                 pt.as_double = data_point.value
                             pb2_metric.gauge.data_points.append(pt)
 
-                    elif isinstance(metric.data, Histogram):
+                    elif isinstance(metric.data, HistogramType):
                         for data_point in metric.data.data_points:
                             pt = pb2.HistogramDataPoint(
                                 attributes=self._translate_attributes(

exporter/opentelemetry-exporter-otlp-proto-grpc/tests/metrics/test_otlp_metrics_exporter.py

@@ -42,10 +42,19 @@
 )
 from opentelemetry.sdk.environment_variables import (
     OTEL_EXPORTER_OTLP_METRICS_INSECURE,
+    OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
 )
-from opentelemetry.sdk.metrics.export import (
-    AggregationTemporality,
+from opentelemetry.sdk.metrics import (
+    Counter,
     Histogram,
+    ObservableCounter,
+    ObservableGauge,
+    ObservableUpDownCounter,
+    UpDownCounter,
+)
+from opentelemetry.sdk.metrics.export import AggregationTemporality
+from opentelemetry.sdk.metrics.export import Histogram as HistogramType
+from opentelemetry.sdk.metrics.export import (
     HistogramDataPoint,
     Metric,
     MetricExportResult,
@@ -121,7 +130,7 @@ def setUp(self):
             name="histogram",
             description="foo",
             unit="s",
-            data=Histogram(
+            data=HistogramType(
                 data_points=[
                     HistogramDataPoint(
                         attributes={"a": 1, "b": True},
@@ -302,6 +311,40 @@ def test_exporting(self):
         # pylint: disable=protected-access
         self.assertEqual(self.exporter._exporting, "metrics")
 
+    @patch.dict(
+        "os.environ",
+        {OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE: "DELTA"},
+    )
+    def test_preferred_temporality(self):
+        # pylint: disable=protected-access
+        exporter = OTLPMetricExporter(
+            preferred_temporality={Counter: AggregationTemporality.CUMULATIVE}
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[Counter],
+            AggregationTemporality.CUMULATIVE,
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[UpDownCounter],
+            AggregationTemporality.CUMULATIVE,
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[Histogram],
+            AggregationTemporality.DELTA,
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[ObservableCounter],
+            AggregationTemporality.DELTA,
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[ObservableUpDownCounter],
+            AggregationTemporality.CUMULATIVE,
+        )
+        self.assertEqual(
+            exporter._preferred_temporality[ObservableGauge],
+            AggregationTemporality.CUMULATIVE,
+        )
+
     @patch(
         "opentelemetry.exporter.otlp.proto.grpc.exporter.ssl_channel_credentials"
     )

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

@@ -31,9 +31,6 @@
     detach,
     set_value,
 )
-from opentelemetry.sdk.environment_variables import (
-    OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
-)
 from opentelemetry.sdk.metrics._internal.aggregation import (
     AggregationTemporality,
     DefaultAggregation,
@@ -73,8 +70,26 @@ class MetricExporter(ABC):
 
     Interface to be implemented by services that want to export metrics received
     in their own format.
+
+    Args:
+        preferred_temporality: Used by `opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader` to
+            configure exporter level preferred temporality. See `opentelemetry.sdk.metrics.export.MetricReader` for
+            more details on what preferred temporality is.
+        preferred_aggregation: Used by `opentelemetry.sdk.metrics.export.PeriodicExportingMetricReader` to
+            configure exporter level preferred aggregation. See `opentelemetry.sdk.metrics.export.MetricReader` for
+            more details on what preferred aggregation is.
     """
 
+    def __init__(
+        self,
+        preferred_temporality: Dict[type, AggregationTemporality] = None,
+        preferred_aggregation: Dict[
+            type, "opentelemetry.sdk.metrics.view.Aggregation"
+        ] = None,
+    ) -> None:
+        self._preferred_temporality = preferred_temporality
+        self._preferred_aggregation = preferred_aggregation
+
     @abstractmethod
     def export(
         self,
@@ -122,6 +137,7 @@ def __init__(
         ] = lambda metrics_data: metrics_data.to_json()
         + linesep,
     ):
+        super().__init__()
         self.out = out
         self.formatter = formatter
 
@@ -143,6 +159,7 @@ def force_flush(self, timeout_millis: float = 10_000) -> bool:
 
 
 class MetricReader(ABC):
+    # pylint: disable=too-many-branches
     """
     Base class for all metric readers
 
@@ -157,8 +174,6 @@ class MetricReader(ABC):
             temporalities of the classes that the user wants to change, not all of
             them. The classes not included in the passed dictionary will retain
             their association to their default aggregation temporalities.
-            The value passed here will override the corresponding values set
-            via the environment variable
         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
@@ -177,10 +192,6 @@ class MetricReader(ABC):
     .. automethod:: _receive_metrics
     """
 
-    # FIXME add :std:envvar:`OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE`
-    # to the end of the documentation paragraph above.
-
-    # pylint: disable=too-many-branches
     def __init__(
         self,
         preferred_temporality: Dict[type, AggregationTemporality] = None,
@@ -196,33 +207,14 @@ def __init__(
             Iterable["opentelemetry.sdk.metrics.export.Metric"],
         ] = None
 
-        if (
-            environ.get(
-                OTEL_EXPORTER_OTLP_METRICS_TEMPORALITY_PREFERENCE,
-                "CUMULATIVE",
-            )
-            .upper()
-            .strip()
-            == "DELTA"
-        ):
-            self._instrument_class_temporality = {
-                _Counter: AggregationTemporality.DELTA,
-                _UpDownCounter: AggregationTemporality.CUMULATIVE,
-                _Histogram: AggregationTemporality.DELTA,
-                _ObservableCounter: AggregationTemporality.DELTA,
-                _ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
-                _ObservableGauge: AggregationTemporality.CUMULATIVE,
-            }
-
-        else:
-            self._instrument_class_temporality = {
-                _Counter: AggregationTemporality.CUMULATIVE,
-                _UpDownCounter: AggregationTemporality.CUMULATIVE,
-                _Histogram: AggregationTemporality.CUMULATIVE,
-                _ObservableCounter: AggregationTemporality.CUMULATIVE,
-                _ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
-                _ObservableGauge: AggregationTemporality.CUMULATIVE,
-            }
+        self._instrument_class_temporality = {
+            _Counter: AggregationTemporality.CUMULATIVE,
+            _UpDownCounter: AggregationTemporality.CUMULATIVE,
+            _Histogram: AggregationTemporality.CUMULATIVE,
+            _ObservableCounter: AggregationTemporality.CUMULATIVE,
+            _ObservableUpDownCounter: AggregationTemporality.CUMULATIVE,
+            _ObservableGauge: AggregationTemporality.CUMULATIVE,
+        }
 
         if preferred_temporality is not None:
             for temporality in preferred_temporality.values():
@@ -404,16 +396,13 @@ class PeriodicExportingMetricReader(MetricReader):
     def __init__(
         self,
         exporter: MetricExporter,
-        preferred_temporality: Dict[type, AggregationTemporality] = None,
-        preferred_aggregation: Dict[
-            type, "opentelemetry.sdk.metrics.view.Aggregation"
-        ] = None,
         export_interval_millis: Optional[float] = None,
         export_timeout_millis: Optional[float] = None,
     ) -> None:
+        # PeriodicExportingMetricReader defers to exporter for configuration
         super().__init__(
-            preferred_temporality=preferred_temporality,
-            preferred_aggregation=preferred_aggregation,
+            preferred_temporality=exporter._preferred_temporality,
+            preferred_aggregation=exporter._preferred_aggregation,
         )
         self._exporter = exporter
         if export_interval_millis is None: