diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi index 5a3ea0ca0b8..3d36dc489d9 100644 --- a/ddtrace/internal/native/_native.pyi +++ b/ddtrace/internal/native/_native.pyi @@ -180,6 +180,15 @@ class SharedRuntime: def shutdown(self, timeout_ms: Optional[int] = None) -> None: """Gracefully shut down the shared runtime. + Args: + timeout_ms: Maximum time in milliseconds to wait for shutdown. + If None, waits indefinitely. + """ + ... + def shutdown_in_thread(self, timeout_ms: Optional[int] = None) -> None: + """Gracefully shut down the shared runtime. + The code is run in a separate thread to bypass a stale thread local storage. + Args: timeout_ms: Maximum time in milliseconds to wait for shutdown. If None, waits indefinitely. diff --git a/ddtrace/internal/native_runtime.py b/ddtrace/internal/native_runtime.py index 969db800fd3..5322dcbbac8 100644 --- a/ddtrace/internal/native_runtime.py +++ b/ddtrace/internal/native_runtime.py @@ -1,4 +1,5 @@ import logging +import sys from typing import Optional from ddtrace.internal import atexit @@ -28,6 +29,7 @@ def __init__(self) -> None: forksafe.register_after_parent(self.after_fork_parent) forksafe.register(self.after_fork_child) atexit.register(self._atexit) + atexit.register_on_exit_signal(self._atexit) def _atexit(self) -> None: try: @@ -43,7 +45,10 @@ def shutdown(self, timeout_ms: Optional[int] = None) -> None: If None, waits indefinitely — only safe if all workers have already been stopped (e.g. via TraceExporter.shutdown). """ - super().shutdown(timeout_ms=timeout_ms) + if "uwsgi" in sys.modules: + super().shutdown_in_thread(timeout_ms=timeout_ms) + else: + super().shutdown(timeout_ms=timeout_ms) atexit.unregister(self._atexit) forksafe.unregister_before_fork(self.before_fork) forksafe.unregister_parent(self.after_fork_parent) diff --git a/ddtrace/internal/writer/writer.py b/ddtrace/internal/writer/writer.py index f1cc6d7a6fc..5d819dedf05 100644 --- a/ddtrace/internal/writer/writer.py +++ b/ddtrace/internal/writer/writer.py @@ -851,7 +851,9 @@ def set_test_session_token(self, token: Optional[str]) -> None: :param token: The test session token to use for authentication. """ self._test_session_token = token + old_exporter = self._exporter self._exporter = self._create_exporter() + old_exporter.shutdown(3_000_000_000) def recreate( self, @@ -866,7 +868,8 @@ def recreate( except ServiceStatusError: # Writers like AgentWriter may not start until the first trace is encoded. # Stopping them before that will raise a ServiceStatusError. - pass + # Shut down the exporter as it's started on init. + self._exporter.shutdown(3_000_000_000) api_version = "v0.4" if (appsec_enabled or llmobs_enabled) else self._api_version return self.__class__( @@ -890,7 +893,9 @@ def _downgrade(self, status, client): if client.ENDPOINT == "v0.5/traces": self._clients = [AgentWriterClientV4(self._buffer_size, self._max_payload_size)] self._api_version = "v0.4" + old_exporter = self._exporter self._exporter = self._create_exporter() + old_exporter.shutdown(3_000_000_000) # Since we have to change the encoding in this case, the payload # would need to be converted to the downgraded encoding before diff --git a/releasenotes/notes/fix-uwsgi-panic-in-worker-8e0cffa85b25592b.yaml b/releasenotes/notes/fix-uwsgi-panic-in-worker-8e0cffa85b25592b.yaml new file mode 100644 index 00000000000..f10178246f6 --- /dev/null +++ b/releasenotes/notes/fix-uwsgi-panic-in-worker-8e0cffa85b25592b.yaml @@ -0,0 +1,3 @@ +fixes: + - | + Fix crashes in uwsgi worker when exiting from SIGTERM. diff --git a/src/native/data_pipeline/mod.rs b/src/native/data_pipeline/mod.rs index 93361c4b8cf..1108476aca7 100644 --- a/src/native/data_pipeline/mod.rs +++ b/src/native/data_pipeline/mod.rs @@ -269,14 +269,6 @@ impl TraceExporterPy { } } -impl Drop for TraceExporterPy { - fn drop(&mut self) { - if let Some(exporter) = self.inner.take() { - let _ = exporter.shutdown(Some(Duration::from_secs(3))); - } - } -} - #[pymodule] pub fn register_data_pipeline(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; diff --git a/src/native/shared_runtime/mod.rs b/src/native/shared_runtime/mod.rs index b1d40cfe568..3e28028c6f6 100644 --- a/src/native/shared_runtime/mod.rs +++ b/src/native/shared_runtime/mod.rs @@ -1,6 +1,7 @@ use libdd_shared_runtime::SharedRuntime; use pyo3::prelude::*; use std::sync::Arc; +use std::thread; use std::time::Duration; mod exceptions; @@ -51,6 +52,19 @@ impl SharedRuntimePy { .map_err(shared_runtime_error_to_pyerr) } + /// Shutdown the runtime in a new thread. + /// This is can be used when thread local storage have been destroyed. + fn shutdown_in_thread(&self, timeout_ms: Option) -> PyResult<()> { + let timeout = timeout_ms.map(Duration::from_millis); + let inner = self.inner.clone(); + let result = thread::spawn(move || inner.shutdown(timeout)) + .join() + .map_err(|_| { + pyo3::exceptions::PyRuntimeError::new_err("Failed to join shutdown thread") + })?; + result.map_err(shared_runtime_error_to_pyerr) + } + fn debug(&self) -> String { format!("{:?}", self.inner) }