Fail any test that leaks a PeriodicExportingMetricReader ticker thread, and fix the four that do

.changelog/5641.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: fail any test that leaves a `PeriodicExportingMetricReader` ticker thread running, and shut down the four readers that were leaking one into the rest of the test session

opentelemetry-sdk/tests/conftest.py

@@ -2,12 +2,26 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import random
+from collections.abc import Iterator
 from os import environ
+from threading import Thread
+from threading import enumerate as enumerate_threads
+from time import monotonic, sleep
 
 import pytest
 
 from opentelemetry.environment_variables import OTEL_PYTHON_CONTEXT
 
+# PeriodicExportingMetricReader names its ticker thread this. The name is
+# hardcoded, so every instance shares it.
+METRIC_READER_THREAD_NAME = "OtelPeriodicExportingMetricReader"
+
+# shutdown() joins the ticker with a timeout, so a thread that is on its way
+# out can still be alive for a moment after the test body returns. Wait this
+# long before calling it a leak. Only a test that actually leaks pays the cost.
+_LEAK_GRACE_PERIOD_SECONDS = 2.0
+_LEAK_POLL_INTERVAL_SECONDS = 0.05
+
 
 def pytest_sessionstart(session):
     # pylint: disable=unused-argument
@@ -23,3 +37,39 @@ def pytest_sessionfinish(session):
 def random_seed():
     # We use random numbers a lot in sampling tests, make sure they are always the same.
     random.seed(0)
+
+
+def live_metric_reader_threads() -> set[Thread]:
+    return {thread for thread in enumerate_threads() if thread.name == METRIC_READER_THREAD_NAME}
+
+
+@pytest.fixture(autouse=True)
+def no_leaked_metric_reader_threads(request: pytest.FixtureRequest) -> Iterator[None]:
+    """Fail a test that leaves a PeriodicExportingMetricReader ticker thread running.
+
+    The ticker is a daemon thread, so nothing reaps it between tests. A leaked
+    one keeps waking up and calling collect() for the rest of the session,
+    inside unrelated tests. That has produced cross-test failures before, see
+    issue #5157.
+
+    Only threads that appear during this test are reported, so one leak fails
+    the test that caused it rather than every test that runs after it.
+    """
+    before = live_metric_reader_threads()
+
+    yield
+
+    leaked = live_metric_reader_threads() - before
+    deadline = monotonic() + _LEAK_GRACE_PERIOD_SECONDS
+    while leaked and monotonic() < deadline:
+        sleep(_LEAK_POLL_INTERVAL_SECONDS)
+        leaked = live_metric_reader_threads() - before
+
+    if leaked:
+        pytest.fail(
+            f"{request.node.nodeid} left {len(leaked)} {METRIC_READER_THREAD_NAME} "
+            "thread(s) running. A leaked ticker thread calls collect() during "
+            "later tests and can fail them (see issue #5157). Shut the reader "
+            "down, or its MeterProvider, from a finally block or addCleanup so "
+            "it happens even when the test fails."
+        )

opentelemetry-sdk/tests/metrics/integration_test/test_exporter_concurrency.py

@@ -79,6 +79,7 @@ def test_exporter_not_called_concurrently(self):
             export_interval_millis=100_000,
         )
         meter_provider = MeterProvider(metric_readers=[reader])
+        self.addCleanup(meter_provider.shutdown)
 
         counter_cb_counter = 0
 

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -101,7 +101,9 @@ def test_register_metric_readers(self):
         mock_exporter._preferred_temporality = None
         mock_exporter._preferred_aggregation = None
         metric_reader_0 = PeriodicExportingMetricReader(mock_exporter)
+        self.addCleanup(metric_reader_0.shutdown)
         metric_reader_1 = PeriodicExportingMetricReader(mock_exporter)
+        self.addCleanup(metric_reader_1.shutdown)
 
         with self.assertNotRaises(Exception):
             MeterProvider(metric_readers=(metric_reader_0,))

opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py

@@ -245,6 +245,7 @@ def test_exporter_temporality_preference(self):
             },
         )
         pmr = PeriodicExportingMetricReader(exporter)
+        self.addCleanup(pmr.shutdown)
         for key, value in pmr._instrument_class_temporality.items():
             if key is not _Counter:
                 self.assertEqual(value, AggregationTemporality.CUMULATIVE)
@@ -258,6 +259,7 @@ def test_exporter_aggregation_preference(self):
             },
         )
         pmr = PeriodicExportingMetricReader(exporter)
+        self.addCleanup(pmr.shutdown)
         for key, value in pmr._instrument_class_aggregation.items():
             if key is not _Counter:
                 self.assertTrue(isinstance(value, DefaultAggregation))