PeriodicExportingMetricReader will continue collection times out (#3098)

In cases where collection times out, the period exporting reader thread should
not terminate, but instead catch, log, and continue on after the regular interval
seconds.

Prior to this commit, a metric collection timeout would terminate the thread and
stop reporting metrics to the wrapped exporter resulting in the appearance in
observability tooling of metrics just stopping without reason.

Author: markallanson

CHANGELOG.md

@@ -28,9 +28,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3156](https://github.com/open-telemetry/opentelemetry-python/pull/3156))
 - deprecate jaeger exporters
   ([#3158](https://github.com/open-telemetry/opentelemetry-python/pull/3158))
-
 - Create a single resource instance
   ([#3118](https://github.com/open-telemetry/opentelemetry-python/pull/3118))
+- PeriodicExportingMetricReader will continue if collection times out 
+  ([#3100](https://github.com/open-telemetry/opentelemetry-python/pull/3100))
+
 
 ## Version 1.15.0/0.36b0 (2022-12-09)
 

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

@@ -41,6 +41,7 @@
     AggregationTemporality,
     DefaultAggregation,
 )
+from opentelemetry.sdk.metrics._internal.exceptions import MetricsTimeoutError
 from opentelemetry.sdk.metrics._internal.instrument import (
     Counter,
     Histogram,
@@ -497,7 +498,14 @@ def _at_fork_reinit(self):
     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)
+            try:
+                self.collect(timeout_millis=self._export_timeout_millis)
+            except MetricsTimeoutError:
+                _logger.warning(
+                    "Metric collection timed out. Will try again after %s seconds",
+                    interval_secs,
+                    exc_info=True,
+                )
         # one last collection below before shutting down completely
         self.collect(timeout_millis=self._export_interval_millis)
 

opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py

@@ -14,12 +14,18 @@
 
 import math
 from time import sleep, time_ns
-from typing import Sequence
+from typing import (
+    Optional,
+    Sequence,
+)
 from unittest.mock import Mock
 
 from flaky import flaky
 
-from opentelemetry.sdk.metrics import Counter
+from opentelemetry.sdk.metrics import (
+    Counter,
+    MetricsTimeoutError,
+)
 from opentelemetry.sdk.metrics._internal import _Counter
 from opentelemetry.sdk.metrics.export import (
     AggregationTemporality,
@@ -67,6 +73,25 @@ def force_flush(self, timeout_millis: float = 10_000) -> bool:
         return True
 
 
+class ExceptionAtCollectionPeriodicExportingMetricReader(
+    PeriodicExportingMetricReader
+):
+    def __init__(
+        self,
+        exporter: MetricExporter,
+        exception: Exception,
+        export_interval_millis: Optional[float] = None,
+        export_timeout_millis: Optional[float] = None,
+    ) -> None:
+        super().__init__(
+            exporter, export_interval_millis, export_timeout_millis
+        )
+        self._collect_exception = exception
+
+    def collect(self, timeout_millis: float = 10_000) -> None:
+        raise self._collect_exception
+
+
 metrics_list = [
     Metric(
         name="sum_name",
@@ -111,11 +136,13 @@ def test_defaults(self):
         pmr.shutdown()
 
     def _create_periodic_reader(
-        self, metrics, exporter, collect_wait=0, interval=60000
+        self, metrics, exporter, collect_wait=0, interval=60000, timeout=30000
     ):
 
         pmr = PeriodicExportingMetricReader(
-            exporter, export_interval_millis=interval
+            exporter,
+            export_interval_millis=interval,
+            export_timeout_millis=timeout,
         )
 
         def _collect(reader, timeout_millis):
@@ -219,3 +246,27 @@ def test_exporter_aggregation_preference(self):
                 self.assertTrue(isinstance(value, DefaultAggregation))
             else:
                 self.assertTrue(isinstance(value, LastValueAggregation))
+
+    def test_metric_timeout_does_not_kill_worker_thread(self):
+        exporter = FakeMetricsExporter()
+        pmr = ExceptionAtCollectionPeriodicExportingMetricReader(
+            exporter,
+            MetricsTimeoutError("test timeout"),
+            export_timeout_millis=1,
+        )
+
+        sleep(0.1)
+        self.assertTrue(pmr._daemon_thread.is_alive())
+        pmr.shutdown()
+
+    def test_non_metric_timeout_error_kills_exporter_thread(self):
+        exporter = FakeMetricsExporter()
+        pmr = ExceptionAtCollectionPeriodicExportingMetricReader(
+            exporter,
+            ZeroDivisionError(),
+            export_timeout_millis=1,
+        )
+
+        sleep(0.1)
+        self.assertFalse(pmr._daemon_thread.is_alive())
+        pmr.shutdown()