fix(signals): only handle signals if in main thread [backport 2.0] (#7218)

Backport 0e21580 from #7204 to 2.0.

Prior to this change ddtrace attempted to register signal handling code
regardless of whether or not it was in the main thread. This caused
ddtrace to fail and exit because in Python signal handling can only
happen in the main thread.

After this change, we only add signal handling capabilities if ddtrace
is imported/initialized as part of the main thread.

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)
- [x] If this PR touches code that signs or publishes builds or
packages, or handles credentials of any kind, I've requested a review
from `@DataDog/security-design-and-guidance`.
- [x] This PR doesn't touch any of that.

---------

Co-authored-by: Teague Bick <[email protected]>

Author: github-actions[bot]

ddtrace/internal/atexit.py

@@ -7,6 +7,7 @@
 import atexit
 import logging
 import signal
+import threading
 import typing
 
 from ddtrace.internal.utils import signals
@@ -66,5 +67,13 @@ def register_on_exit_signal(f):
     def handle_exit(sig, frame):
         f()
 
-    signals.handle_signal(signal.SIGTERM, handle_exit)
-    signals.handle_signal(signal.SIGINT, handle_exit)
+    if threading.current_thread() is threading.main_thread():
+        try:
+            signals.handle_signal(signal.SIGTERM, handle_exit)
+            signals.handle_signal(signal.SIGINT, handle_exit)
+        except Exception:
+            # We catch a general exception here because we don't know
+            # what might go wrong, but we don't want to stop
+            # normal program execution based upon failing to register
+            # a signal handler.
+            log.debug("Encountered an exception while registering a signal", exc_info=True)

releasenotes/notes/signals-fix-import-in-thread-error-1d68cb76472268d7.yaml

@@ -0,0 +1,3 @@
+fixes:
+  - |
+    data_streams: This fix resolves an issue where including exit signal handling for message queues caused imports of ddtrace in a thread to fail.

tests/datastreams/test_processor.py

@@ -97,3 +97,29 @@ def run_test():
     env["DD_DATA_STREAMS_ENABLED"] = "True"
     out, err, status, _ = ddtrace_run_python_code_in_subprocess(code, env=env, timeout=5)
     assert out.decode().strip() == "Fake flush called"
+
+
+def test_threaded_import(ddtrace_run_python_code_in_subprocess):
+    code = """
+import pytest
+import sys
+import time
+import threading
+
+from ddtrace.internal.datastreams.processor import DataStreamsProcessor
+
+def fake_flush(*args, **kwargs):
+    print("Fake flush called")
+
+def run_test():
+    processor = DataStreamsProcessor("http://localhost:8126")
+
+t = threading.Thread(target=run_test)
+t.start()
+t.join()
+"""
+
+    env = os.environ.copy()
+    env["DD_DATA_STREAMS_ENABLED"] = "True"
+    out, err, status, _ = ddtrace_run_python_code_in_subprocess(code, env=env, timeout=5)
+    assert err.decode().strip() == ""

tests/tracer/test_tracer.py

@@ -1129,6 +1129,18 @@ def test_enable():
         assert not t2.enabled
 
 
+@pytest.mark.subprocess(parametrize={"DD_TRACE_ENABLED": ["true", "false"]})
+def test_threaded_import():
+    import threading
+
+    def thread_target():
+        import ddtrace  # noqa: F401
+
+    t = threading.Thread(target=thread_target)
+    t.start()
+    t.join()
+
+
 def test_runtime_id_parent_only():
     tracer = ddtrace.Tracer()