diff --git a/.changelog/5564.fixed b/.changelog/5564.fixed new file mode 100644 index 00000000000..eed3be984be --- /dev/null +++ b/.changelog/5564.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: pass the remaining shutdown budget to the metric exporter as `timeout_millis`, so `PeriodicExportingMetricReader.shutdown` no longer silently falls back to the exporter default diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py index cb1040537d3..3f5978a030b 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -538,7 +538,10 @@ def _shutdown(): self._shutdown_event.set() if self._daemon_thread: self._daemon_thread.join(timeout=(deadline_ns - time_ns()) / 10**9) - self._exporter.shutdown(timeout=(deadline_ns - time_ns()) / 10**6) + # `timeout_millis` is the parameter every MetricExporter.shutdown + # declares; anything else is swallowed by its **kwargs. Clamp, because + # joining the ticker thread above may already have spent the budget. + self._exporter.shutdown(timeout_millis=max(0, (deadline_ns - time_ns()) / 10**6)) def force_flush(self, timeout_millis: float = 10_000) -> bool: super().force_flush(timeout_millis=timeout_millis) diff --git a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py index 76ad37b8dbc..a2c8048662a 100644 --- a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py +++ b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py @@ -11,6 +11,7 @@ from logging import WARNING from time import sleep, time_ns from typing import cast +from unittest import TestCase from unittest.mock import Mock, patch import pytest @@ -395,3 +396,64 @@ def test_metric_reader_metrics(self): self.assertTrue(name.startswith("periodic_metric_reader/")) mp.shutdown() + + +class TestPeriodicExportingMetricReaderShutdownTimeout(TestCase): + """The reader must hand the remaining budget to the exporter. + + Every MetricExporter.shutdown signature is + ``shutdown(self, timeout_millis=30_000, **kwargs)``, so passing the budget + under any other keyword silently lands in **kwargs and the exporter falls + back to its own 30s default. + """ + + class _RecordingExporter(MetricExporter): + def __init__(self): + super().__init__( + preferred_temporality={Counter: AggregationTemporality.CUMULATIVE}, + ) + self.shutdown_timeout_millis = None + self.shutdown_kwargs = None + + def export(self, metrics_data, timeout_millis=10_000, **kwargs): + return MetricExportResult.SUCCESS + + def force_flush(self, timeout_millis=10_000): + return True + + def shutdown(self, timeout_millis: float = 30_000, **kwargs) -> None: + self.shutdown_timeout_millis = timeout_millis + self.shutdown_kwargs = kwargs + + def _shutdown_with(self, timeout_millis): + exporter = self._RecordingExporter() + reader = PeriodicExportingMetricReader(exporter, export_interval_millis=600_000) + MeterProvider(metric_readers=[reader]).shutdown(timeout_millis=timeout_millis) + return exporter + + def test_exporter_receives_remaining_budget_as_timeout_millis(self): + exporter = self._shutdown_with(500) + self.assertIsNotNone(exporter.shutdown_timeout_millis) + # some of the budget is consumed joining the ticker thread + self.assertLessEqual(exporter.shutdown_timeout_millis, 500) + self.assertGreater(exporter.shutdown_timeout_millis, 0) + + def test_exporter_does_not_receive_the_budget_as_an_unknown_kwarg(self): + exporter = self._shutdown_with(500) + self.assertEqual(exporter.shutdown_kwargs, {}) + + def test_exporter_does_not_fall_back_to_its_own_default(self): + exporter = self._shutdown_with(500) + self.assertNotEqual(exporter.shutdown_timeout_millis, 30_000) + + def test_budget_is_never_negative(self): + """An already-exhausted budget must clamp, not go negative. + + Driven through the reader directly: MeterProvider.shutdown enforces its + own deadline and would refuse to call the reader at all. + """ + exporter = self._RecordingExporter() + reader = PeriodicExportingMetricReader(exporter, export_interval_millis=600_000) + MeterProvider(metric_readers=[reader]) + reader.shutdown(timeout_millis=0) + self.assertGreaterEqual(exporter.shutdown_timeout_millis, 0)