fix: reinitialize locks after fork to prevent deadlocks in child processes

Summary
This commit adds post-fork reinitialization of threading locks across multiple components in the OpenTelemetry Python SDK and API. It ensures that threading.Lock() instances are safely reinitialized in child processes after a fork(), preventing potential deadlocks and undefined behavior.

Details
Introduced usage of register_at_fork(after_in_child=...) from the os module to reinitialize thread locks.

Used weakref.WeakMethod() to safely refer to bound instance methods in register_at_fork.

Added _at_fork_reinit() methods to classes using threading locks and registered them to run in child processes post-fork.

Applied this to all usages of Lock, RLock

Rationale
Forked child processes inherit thread state from the parent, including the internal state of locks. This can cause deadlocks or runtime errors if a lock was held at the time of the fork. By reinitializing locks using the register_at_fork mechanism, we ensure child processes start with clean lock states.

This is especially relevant for WSGI servers and environments that use pre-fork models (e.g., gunicorn, uWSGI), where instrumentation and telemetry components may misbehave without this precaution.

Author: dshivashankar1994

exporter/opentelemetry-exporter-otlp-proto-grpc/src/opentelemetry/exporter/otlp/proto/grpc/exporter.py

@@ -15,10 +15,11 @@
 """OTLP Exporter"""
 
 import threading
+import weakref
 from abc import ABC, abstractmethod
 from collections.abc import Sequence  # noqa: F401
 from logging import getLogger
-from os import environ
+from os import environ, register_at_fork
 from time import sleep
 from typing import (  # noqa: F401
     Any,
@@ -261,8 +262,13 @@ def __init__(
         self._client = self._stub(self._channel)
 
         self._export_lock = threading.Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
         self._shutdown = False
 
+    def _at_fork_reinit(self):
+        self._export_lock._at_fork_reinit()
+
     @abstractmethod
     def _translate_data(
         self, data: TypingSequence[SDKDataT]

opentelemetry-api/src/opentelemetry/attributes/__init__.py

@@ -14,9 +14,11 @@
 
 import logging
 import threading
+import weakref
 from collections import OrderedDict
 from collections.abc import MutableMapping
 from typing import Mapping, Optional, Sequence, Tuple, Union
+from os import register_at_fork
 
 from opentelemetry.util import types
 
@@ -263,6 +265,11 @@ def __init__(
             for key, value in attributes.items():
                 self[key] = value
         self._immutable = immutable
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def __repr__(self) -> str:
         return f"{dict(self._dict)}"

opentelemetry-api/src/opentelemetry/metrics/_internal/__init__.py

@@ -41,10 +41,11 @@
 """
 
 import warnings
+import weakref
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from logging import getLogger
-from os import environ
+from os import environ, register_at_fork
 from threading import Lock
 from typing import Dict, List, Optional, Sequence, Union, cast
 
@@ -156,6 +157,11 @@ def __init__(self) -> None:
         self._lock = Lock()
         self._meters: List[_ProxyMeter] = []
         self._real_meter_provider: Optional[MeterProvider] = None
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def get_meter(
         self,
@@ -510,6 +516,11 @@ def __init__(
         self._lock = Lock()
         self._instruments: List[_ProxyInstrumentT] = []
         self._real_meter: Optional[Meter] = None
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def on_set_meter_provider(self, meter_provider: MeterProvider) -> None:
         """Called when a real meter provider is set on the creating _ProxyMeterProvider

opentelemetry-api/src/opentelemetry/util/_once.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import weakref
+from os import register_at_fork
 from threading import Lock
 from typing import Callable
 
@@ -25,6 +27,11 @@ class Once:
     def __init__(self) -> None:
         self._lock = Lock()
         self._done = False
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def do_once(self, func: Callable[[], None]) -> bool:
         """Execute ``func`` if it hasn't been executed or return.

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/__init__.py

@@ -21,7 +21,8 @@
 import threading
 import traceback
 import warnings
-from os import environ
+import weakref
+from os import environ, register_at_fork
 from threading import Lock
 from time import time_ns
 from typing import Any, Callable, Tuple, Union, cast  # noqa
@@ -317,6 +318,11 @@ def __init__(self):
         # iterating through it on "emit".
         self._log_record_processors = ()  # type: Tuple[LogRecordProcessor, ...]
         self._lock = threading.Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def add_log_record_processor(
         self, log_record_processor: LogRecordProcessor
@@ -379,6 +385,11 @@ def __init__(self, max_workers: int = 2):
         self._executor = concurrent.futures.ThreadPoolExecutor(
             max_workers=max_workers
         )
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
 
     def add_log_record_processor(
         self, log_record_processor: LogRecordProcessor
@@ -633,6 +644,11 @@ def __init__(
             self._at_exit_handler = atexit.register(self.shutdown)
         self._logger_cache = {}
         self._logger_cache_lock = Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._logger_cache_lock._at_fork_reinit()
 
     @property
     def resource(self):

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/in_memory_log_exporter.py

@@ -14,6 +14,8 @@
 
 import threading
 import typing
+import weakref
+from os import register_at_fork
 
 from opentelemetry.sdk._logs import LogData
 from opentelemetry.sdk._logs.export import LogExporter, LogExportResult
@@ -30,8 +32,13 @@ class InMemoryLogExporter(LogExporter):
     def __init__(self):
         self._logs = []
         self._lock = threading.Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
         self._stopped = False
 
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
+
     def clear(self) -> None:
         with self._lock:
             self._logs.clear()

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

@@ -15,7 +15,7 @@
 import weakref
 from atexit import register, unregister
 from logging import getLogger
-from os import environ
+from os import environ, register_at_fork
 from threading import Lock
 from time import time_ns
 from typing import Optional, Sequence
@@ -88,6 +88,11 @@ def __init__(
         self._measurement_consumer = measurement_consumer
         self._instrument_id_instrument = {}
         self._instrument_id_instrument_lock = Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
+
+    def _at_fork_reinit(self):
+        self._instrument_id_instrument_lock._at_fork_reinit()
 
     def create_counter(self, name, unit="", description="") -> APICounter:
         status = self._register_instrument(name, _Counter, unit, description)
@@ -421,6 +426,8 @@ def __init__(
     ):
         self._lock = Lock()
         self._meter_lock = Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
         self._atexit_handler = None
         if resource is None:
             resource = Resource.create({})
@@ -463,6 +470,14 @@ def __init__(
                 self._measurement_consumer.collect
             )
 
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
+        self._meter_lock._at_fork_reinit()
+
+    @classmethod
+    def _register_fork_handlers(cls):
+        register_at_fork(after_in_child=cls._all_metric_readers_lock._at_fork_reinit)
+
     def force_flush(self, timeout_millis: float = 10_000) -> bool:
         deadline_ns = time_ns() + timeout_millis * 10**6
 
@@ -580,3 +595,6 @@ def get_meter(
                     self._measurement_consumer,
                 )
             return self._meters[info]
+
+# Call the method after the class is fully defined
+MeterProvider._register_fork_handlers()

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

@@ -13,7 +13,9 @@
 # limitations under the License.
 
 
+import weakref
 from logging import getLogger
+from os import register_at_fork
 from threading import Lock
 from time import time_ns
 from typing import Dict, List, Optional, Sequence
@@ -44,6 +46,8 @@ def __init__(
         self._instrument = instrument
         self._attributes_aggregation: Dict[frozenset, _Aggregation] = {}
         self._lock = Lock()
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
         self._instrument_class_aggregation = instrument_class_aggregation
         self._name = self._view._name or self._instrument.name
         self._description = (
@@ -66,6 +70,9 @@ def __init__(
                 0,
             )
 
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
+
     def conflicts(self, other: "_ViewInstrumentMatch") -> bool:
         # pylint: disable=protected-access
 

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

@@ -14,12 +14,14 @@
 
 # pylint: disable=too-many-lines
 
+import weakref
 from abc import ABC, abstractmethod
 from bisect import bisect_left
 from enum import IntEnum
 from functools import partial
 from logging import getLogger
 from math import inf
+from os import register_at_fork
 from threading import Lock
 from typing import (
     Callable,
@@ -98,9 +100,14 @@ def __init__(
     ):
         self._lock = Lock()
         self._attributes = attributes
+        weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
+        register_at_fork(after_in_child=lambda: weak_reinit()())
         self._reservoir = reservoir_builder()
         self._previous_point = None
 
+    def _at_fork_reinit(self):
+        self._lock._at_fork_reinit()
+
     @abstractmethod
     def aggregate(
         self, measurement: Measurement, should_sample_exemplar: bool = True

opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/exponential_histogram/mapping/exponent_mapping.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from math import ldexp
+from os import register_at_fork
 from threading import Lock
 
 from opentelemetry.sdk.metrics._internal.exponential_histogram.mapping import (
@@ -42,6 +43,11 @@ class ExponentMapping(Mapping):
     _min_scale = -10
     _max_scale = 0
 
+    # Add a class method for initialization that includes fork handler registration
+    @classmethod
+    def _register_fork_handlers(cls):
+        register_at_fork(after_in_child=cls._mappings_lock._at_fork_reinit)
+
     def _get_min_scale(self):
         # _min_scale defines the point at which the exponential mapping
         # function becomes useless for 64-bit floats. With scale -10, ignoring
@@ -139,3 +145,6 @@ def get_lower_boundary(self, index: int) -> float:
     @property
     def scale(self) -> int:
         return self._scale
+
+# Call the method after the class is fully defined
+ExponentMapping._register_fork_handlers()