PeriodicExportingMetricsReader with value = infinity to support explicit metric collection

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

@@ -411,22 +411,17 @@ def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
         pass
 
 
-class PeriodicExportingMetricReader(MetricReader):
-    """`PeriodicExportingMetricReader` is an implementation of `MetricReader`
-    that collects metrics based on a user-configurable time interval, and passes the
-    metrics to the configured exporter.
-
-    The configured exporter's :py:meth:`~MetricExporter.export` method will not be called
-    concurrently.
+class ExportingMetricReader(MetricReader):
+    """`ExportingMetricReader` is an implementation of `MetricReader`
+    that collects metrics and passes the metrics to the configured exporter, explicitly.
     """
 
     def __init__(
         self,
         exporter: MetricExporter,
-        export_interval_millis: Optional[float] = None,
         export_timeout_millis: Optional[float] = None,
     ) -> None:
-        # PeriodicExportingMetricReader defers to exporter for configuration
+        # ExportingMetricReader defers to exporter for configuration
         super().__init__(
             preferred_temporality=exporter._preferred_temporality,
             preferred_aggregation=exporter._preferred_aggregation,
@@ -438,16 +433,7 @@ def __init__(
         self._export_lock = Lock()
 
         self._exporter = exporter
-        if export_interval_millis is None:
-            try:
-                export_interval_millis = float(
-                    environ.get(OTEL_METRIC_EXPORT_INTERVAL, 60000)
-                )
-            except ValueError:
-                _logger.warning(
-                    "Found invalid value for export interval, using default"
-                )
-                export_interval_millis = 60000
+
         if export_timeout_millis is None:
             try:
                 export_timeout_millis = float(
@@ -458,36 +444,15 @@ def __init__(
                     "Found invalid value for export timeout, using default"
                 )
                 export_timeout_millis = 30000
-        self._export_interval_millis = export_interval_millis
+
         self._export_timeout_millis = export_timeout_millis
         self._shutdown = False
         self._shutdown_event = Event()
         self._shutdown_once = Once()
-        self._daemon_thread = Thread(
-            name="OtelPeriodicExportingMetricReader",
-            target=self._ticker,
-            daemon=True,
-        )
-        self._daemon_thread.start()
-        if hasattr(os, "register_at_fork"):
-            os.register_at_fork(
-                after_in_child=self._at_fork_reinit
-            )  # pylint: disable=protected-access
 
-    def _at_fork_reinit(self):
-        self._daemon_thread = Thread(
-            name="OtelPeriodicExportingMetricReader",
-            target=self._ticker,
-            daemon=True,
-        )
-        self._daemon_thread.start()
-
-    def _ticker(self) -> None:
-        interval_secs = self._export_interval_millis / 1e3
-        while not self._shutdown_event.wait(interval_secs):
-            self.collect(timeout_millis=self._export_timeout_millis)
-        # one last collection below before shutting down completely
-        self.collect(timeout_millis=self._export_interval_millis)
+    def export(self, timeout_millis: float = 10_000) -> None:
+        # export is just a wrapper on force_flush.
+        return self.force_flush(timeout_millis=timeout_millis)
 
     def _receive_metrics(
         self,
@@ -519,10 +484,74 @@ def _shutdown():
             return
 
         self._shutdown_event.set()
-        self._daemon_thread.join(timeout=(deadline_ns - time_ns()) / 10**9)
         self._exporter.shutdown(timeout=(deadline_ns - time_ns()) / 10**6)
 
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
         super().force_flush(timeout_millis=timeout_millis)
         self._exporter.force_flush(timeout_millis=timeout_millis)
         return True
+
+
+class PeriodicExportingMetricReader(ExportingMetricReader):
+    """`PeriodicExportingMetricReader` is an implementation of `ExportingMetricReader`
+    that collects metrics based on a user-configurable time interval, and passes the
+    metrics to the configured exporter.
+
+    The configured exporter's :py:meth:`~MetricExporter.export` method will not be called
+    concurrently.
+    """
+
+    def __init__(
+        self,
+        exporter: MetricExporter,
+        export_interval_millis: Optional[float] = None,
+        export_timeout_millis: Optional[float] = None,
+    ) -> None:
+        # PeriodicExportingMetricReader defers to exporter for configuration
+        super().__init__(
+            exporter=exporter,
+            export_timeout_millis=export_timeout_millis,
+        )
+
+        if export_interval_millis is None:
+            try:
+                export_interval_millis = float(
+                    environ.get(OTEL_METRIC_EXPORT_INTERVAL, 60000)
+                )
+            except ValueError:
+                _logger.warning(
+                    "Found invalid value for export interval, using default"
+                )
+                export_interval_millis = 60000
+
+        self._export_interval_millis = export_interval_millis
+        self._daemon_thread = Thread(
+            name="OtelPeriodicExportingMetricReader",
+            target=self._ticker,
+            daemon=True,
+        )
+        self._daemon_thread.start()
+        if hasattr(os, "register_at_fork"):
+            os.register_at_fork(
+                after_in_child=self._at_fork_reinit
+            )  # pylint: disable=protected-access
+
+    def _at_fork_reinit(self):
+        self._daemon_thread = Thread(
+            name="OtelPeriodicExportingMetricReader",
+            target=self._ticker,
+            daemon=True,
+        )
+        self._daemon_thread.start()
+
+    def _ticker(self) -> None:
+        interval_secs = self._export_interval_millis / 1e3
+        while not self._shutdown_event.wait(interval_secs):
+            self.collect(timeout_millis=self._export_timeout_millis)
+        # one last collection below before shutting down completely
+        self.collect(timeout_millis=self._export_interval_millis)
+
+    def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
+        deadline_ns = time_ns() + timeout_millis * 10**6
+        super().shutdown(timeout_millis=timeout_millis)
+        self._daemon_thread.join(timeout=(deadline_ns - time_ns()) / 10**9)

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

@@ -17,6 +17,7 @@
 from opentelemetry.sdk.metrics._internal.export import (  # noqa: F401
     AggregationTemporality,
     ConsoleMetricExporter,
+    ExportingMetricReader,
     InMemoryMetricReader,
     MetricExporter,
     MetricExportResult,

opentelemetry-sdk/tests/metrics/test_exporting_metric_reader.py

@@ -0,0 +1,185 @@
+# 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 time import sleep, time_ns
+from typing import Sequence
+from unittest.mock import Mock
+
+from flaky import flaky
+
+from opentelemetry.sdk.metrics import Counter
+from opentelemetry.sdk.metrics._internal import _Counter
+from opentelemetry.sdk.metrics.export import (
+    AggregationTemporality,
+    ExportingMetricReader,
+    Gauge,
+    Metric,
+    MetricExporter,
+    MetricExportResult,
+    NumberDataPoint,
+    Sum,
+)
+from opentelemetry.sdk.metrics.view import (
+    DefaultAggregation,
+    LastValueAggregation,
+)
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
+
+
+class FakeMetricsExporter(MetricExporter):
+    def __init__(
+        self, wait=0, preferred_temporality=None, preferred_aggregation=None
+    ):
+        self.wait = wait
+        self.metrics = []
+        self._shutdown = False
+        super().__init__(
+            preferred_temporality=preferred_temporality,
+            preferred_aggregation=preferred_aggregation,
+        )
+
+    def export(
+        self,
+        metrics: Sequence[Metric],
+        timeout_millis: float = 10_000,
+        **kwargs,
+    ) -> MetricExportResult:
+        sleep(self.wait)
+        self.metrics.extend(metrics)
+        return True
+
+    def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None:
+        self._shutdown = True
+
+    def force_flush(self, timeout_millis: float = 10_000) -> bool:
+        return True
+
+
+metrics_list = [
+    Metric(
+        name="sum_name",
+        description="",
+        unit="",
+        data=Sum(
+            data_points=[
+                NumberDataPoint(
+                    attributes={},
+                    start_time_unix_nano=time_ns(),
+                    time_unix_nano=time_ns(),
+                    value=2,
+                )
+            ],
+            aggregation_temporality=1,
+            is_monotonic=True,
+        ),
+    ),
+    Metric(
+        name="gauge_name",
+        description="",
+        unit="",
+        data=Gauge(
+            data_points=[
+                NumberDataPoint(
+                    attributes={},
+                    start_time_unix_nano=time_ns(),
+                    time_unix_nano=time_ns(),
+                    value=2,
+                )
+            ]
+        ),
+    ),
+]
+
+
+class TestExportingMetricReader(ConcurrencyTestBase):
+    def test_defaults(self):
+        pmr = ExportingMetricReader(FakeMetricsExporter())
+        self.assertEqual(pmr._export_timeout_millis, 30000)
+        pmr.shutdown()
+
+    def _create_exporting_reader(
+        self, metrics, exporter, collect_wait=0, timeout=30000
+    ):
+
+        pmr = ExportingMetricReader(exporter, export_timeout_millis=timeout)
+
+        def _collect(reader, timeout_millis):
+            sleep(collect_wait)
+            pmr._receive_metrics(metrics, timeout_millis)
+
+        pmr._set_collect_callback(_collect)
+        return pmr
+
+    def test_export(self):
+        collect_mock = Mock()
+        exporter = FakeMetricsExporter()
+        exporter.export = Mock()
+        pmr = ExportingMetricReader(exporter, export_timeout_millis=1)
+        pmr._set_collect_callback(collect_mock)
+        pmr.export()
+        self.assertTrue(collect_mock.assert_called_once)
+        pmr.shutdown()
+
+    @flaky(max_runs=3, min_passes=1)
+    def test_collects_metrics(self):
+        exporter = FakeMetricsExporter()
+
+        pmr = self._create_exporting_reader(
+            metrics_list, exporter, timeout=1000
+        )
+        pmr.collect()
+        self.assertEqual(exporter.metrics, metrics_list)
+        pmr.shutdown()
+
+    def test_shutdown(self):
+        exporter = FakeMetricsExporter()
+
+        pmr = self._create_exporting_reader([], exporter)
+        pmr.shutdown()
+        self.assertEqual(exporter.metrics, [])
+        self.assertTrue(pmr._shutdown)
+        self.assertTrue(exporter._shutdown)
+
+    def test_shutdown_multiple_times(self):
+        pmr = self._create_exporting_reader([], FakeMetricsExporter())
+        with self.assertLogs(level="WARNING") as w:
+            self.run_with_many_threads(pmr.shutdown)
+            self.assertTrue("Can't shutdown multiple times", w.output[0])
+        pmr.shutdown()
+
+    def test_exporter_temporality_preference(self):
+        exporter = FakeMetricsExporter(
+            preferred_temporality={
+                Counter: AggregationTemporality.DELTA,
+            },
+        )
+        pmr = ExportingMetricReader(exporter)
+        for key, value in pmr._instrument_class_temporality.items():
+            if key is not _Counter:
+                self.assertEqual(value, AggregationTemporality.CUMULATIVE)
+            else:
+                self.assertEqual(value, AggregationTemporality.DELTA)
+
+    def test_exporter_aggregation_preference(self):
+        exporter = FakeMetricsExporter(
+            preferred_aggregation={
+                Counter: LastValueAggregation(),
+            },
+        )
+        pmr = ExportingMetricReader(exporter)
+        for key, value in pmr._instrument_class_aggregation.items():
+            if key is not _Counter:
+                self.assertTrue(isinstance(value, DefaultAggregation))
+            else:
+                self.assertTrue(isinstance(value, LastValueAggregation))