fix(metrics): Add fork-safety to SynchronousMeasurementConsumer

.changelog/4767.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: Add fork-safety to metrics `SynchronousMeasurementConsumer` by registering a post-fork child hook, lazily reinitializing metric reader storages on first use in the child process, and clearing asynchronous instruments to avoid duplicated state after `fork()`

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/measurement_consumer.py

@@ -3,6 +3,7 @@
 
 # pylint: disable=unused-import
 
+import os
 import weakref
 from abc import ABC, abstractmethod
 from collections.abc import Iterable, Mapping
@@ -65,8 +66,30 @@ def __init__(
         self._async_instruments: list[
             opentelemetry.sdk.metrics._internal.instrument._Asynchronous
         ] = []
+        self._needs_storage_reinit = False
+        if hasattr(os, "register_at_fork"):
+            weak_self = weakref.ref(self)
+
+            def _fork_handler():
+                obj = weak_self()
+                if obj is not None:
+                    obj._at_fork_reinit()  # pylint: disable=protected-access
+
+            os.register_at_fork(after_in_child=_fork_handler)
+
+    def _at_fork_reinit(self):
+        """Reinitialize lock in child process after fork"""
+        self._lock = Lock()
+        # Lazy reinitialization of storages on first use post fork. This is
+        # done to avoid the overhead of reinitializing the storages on
+        # every fork.
+        self._needs_storage_reinit = True
+        self._async_instruments.clear()
 
     def consume_measurement(self, measurement: Measurement) -> None:
+        if self._needs_storage_reinit:
+            self._reinit_storages()
+
         should_sample_exemplar = (
             self._sdk_config.exemplar_filter.should_sample(
                 measurement.value,
@@ -96,6 +119,9 @@ def collect(
         metric_reader: "opentelemetry.sdk.metrics.export.MetricReader",
         timeout_millis: float = 10_000,
     ) -> MetricsData | None:
+        if self._needs_storage_reinit:
+            self._reinit_storages()
+
         with self._lock:
             metric_reader_storage = self._reader_storages[metric_reader]
             # for now, just use the defaults
@@ -135,6 +161,17 @@ def collect(
 
         return result
 
+    def _reinit_storages(self):
+        # Reinitialize the storages after a fork to avoid duplicate data points.
+        # The flag is cleared inside the lock so concurrent callers only run
+        # reinit once.
+        with self._lock:
+            if not self._needs_storage_reinit:
+                return
+            for storage in self._reader_storages.values():
+                storage._at_fork_reinit()  # pylint: disable=protected-access
+            self._needs_storage_reinit = False
+
     def add_metric_reader(
         self, metric_reader: "opentelemetry.sdk.metrics.MetricReader"
     ) -> None:

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/metric_reader_storage.py

@@ -63,6 +63,11 @@ def __init__(
         self._instrument_class_temporality = instrument_class_temporality
         self._instrument_class_aggregation = instrument_class_aggregation
 
+    def _at_fork_reinit(self) -> None:
+        """Reinitialize locks and clear state in the child process after fork."""
+        self._lock = RLock()
+        self._instrument_view_instrument_matches.clear()
+
     def _get_or_init_view_instrument_match(
         self, instrument: _Instrument
     ) -> list[_ViewInstrumentMatch]: