fix(tracer): avoid flushing traces from the child in case of fork (#18262) (#18646)

Fixes duplicate trace emission after `fork()` when a trace is already
buffered in the parent writer.

My understanding of the problem is that in the child after a fork,
`tracer._child_after_fork` recreates the writer and in the process
properly shuts down its worker thread which calls the `on_shutdown`
callback that flushes the traces from the child. The same traces are
also emitted periodically from the parent and it causes the double trace
emission problem.

A solution is to skip restarting the worker thread in the writer. This
way when the tracer is recreated in `tracer._child_after_fork` the
worker thread is not stopped a second time and does not flush traces.

When the thread is stopped for the first time in
`PeriodicThread_after_fork`, the `_skip_shutdown` parameter is set to
`false` so flushing does not occur yet.

We noticed the problem while introducing Stripe tests in the
system-tests "DEFAULT" scenario. The `stripe` SDK forks on requests and
this causes a duplication of a few traces preceding that stripe test.
This is caught by an integrity test check that runs on all traces during
the scenario. The exact problematic lines in the stripe SDK are:
https://github.com/stripe/stripe-python/blob/v12.5.1/stripe/_api_requestor.py#L475

In general `subprocess.Popen` (used to exec `uname` here) should perform
an `os.posix_spawn` and not trigger fork hooks. However, gevent patches
`subprocess.Popen` to use `os.fork()` instead.

- Added a regression test


(cherry picked from commit aa02e52)

Co-authored-by: Florentin Labelle <florentin.labelle@datadoghq.com>

Author: brettlangdon

ddtrace/internal/writer/writer.py

@@ -225,7 +225,7 @@ def __init__(
             processing_interval = config._trace_writer_interval_seconds
         if timeout is None:
             timeout = agent_config.trace_agent_timeout_seconds
-        super(HTTPWriter, self).__init__(interval=processing_interval)
+        super(HTTPWriter, self).__init__(interval=processing_interval, autorestart=False)
         self.intake_url = intake_url
         self._intake_accepts_gzip = use_gzip
         self._buffer_size = buffer_size
@@ -707,7 +707,7 @@ def __init__(
             if "X-Datadog-Test-Session-Token" in additional_header:
                 test_session_token = additional_header["X-Datadog-Test-Session-Token"]
 
-        super(NativeWriter, self).__init__(interval=processing_interval)
+        super(NativeWriter, self).__init__(interval=processing_interval, autorestart=False)
         self.intake_url = intake_url
         self._otlp_endpoint = otlp_endpoint
         self._buffer_size = buffer_size

releasenotes/notes/fix-duplicate-traces-after-fork-15f1edf0a3445a81.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    tracing: This fix resolves an issue where traces buffered before ``os.fork()`` could be sent twice, once by the parent and once by the child.

tests/integration/test_integration_snapshots.py

@@ -501,3 +501,44 @@ def test_signal_shutdown_flushes_traces(
             if proc.poll() is None:
                 proc.kill()
                 proc.wait()
+
+
+@pytest.mark.subprocess(err=None, env={"DD_TRACE_WRITER_INTERVAL_SECONDS": "30"})
+@pytest.mark.snapshot()
+def test_buffered_trace_not_duplicated_across_fork():
+    """A trace buffered in the parent before fork() must be sent exactly once."""
+    import os
+
+    from ddtrace.trace import tracer
+
+    with tracer.trace("buffered-before-fork", service="fork-test"):
+        pass
+
+    pid = os.fork()
+    if pid == 0:
+        os._exit(0)
+
+    os.waitpid(pid, 0)
+    tracer.flush()
+
+
+@pytest.mark.subprocess(err=None, env={"DD_TRACE_WRITER_INTERVAL_SECONDS": "30"})
+@pytest.mark.snapshot()
+def test_writer_restarts_in_child_and_flushes_traces_after_fork():
+    """After fork the child can create and flush its own traces while the parent trace is sent exactly once."""
+    import os
+
+    from ddtrace.trace import tracer
+
+    with tracer.trace("parent-span", service="fork-test"):
+        pass
+
+    pid = os.fork()
+    if pid == 0:
+        with tracer.trace("child-span", service="fork-test"):
+            pass
+        tracer.flush()
+        os._exit(0)
+
+    os.waitpid(pid, 0)
+    tracer.flush()

tests/snapshots/tests.integration.test_integration_snapshots.test_buffered_trace_not_duplicated_across_fork.json

@@ -0,0 +1,28 @@
+[[
+  {
+    "name": "buffered-before-fork",
+    "service": "fork-test",
+    "resource": "buffered-before-fork",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.base_service": "ddtrace_subprocess_dir",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6a27e37000000000",
+      "_dd.svc_src": "m",
+      "_dd.tags.process": "entrypoint.basedir:ddtrace_subprocess_dir,entrypoint.name:tmpfe9lhcj6,entrypoint.type:script,entrypoint.workdir:dd-trace-py,svc.auto:ddtrace_subprocess_dir",
+      "language": "python",
+      "runtime-id": "47d2ebc065d54d29881077fd2cb54d21"
+    },
+    "metrics": {
+      "_dd.top_level": 1.0,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1.0,
+      "process_id": 91747.0
+    },
+    "duration": 2233000,
+    "start": 1780999024913878000
+  }]]

tests/snapshots/tests.integration.test_integration_snapshots.test_writer_restarts_in_child_and_flushes_traces_after_fork.json

@@ -0,0 +1,56 @@
+[[
+  {
+    "name": "parent-span",
+    "service": "fork-test",
+    "resource": "parent-span",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.base_service": "ddtrace_subprocess_dir",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6a27e35500000000",
+      "_dd.svc_src": "m",
+      "_dd.tags.process": "entrypoint.basedir:ddtrace_subprocess_dir,entrypoint.name:tmpuq52vj3g,entrypoint.type:script,entrypoint.workdir:dd-trace-py,svc.auto:ddtrace_subprocess_dir",
+      "language": "python",
+      "runtime-id": "e9137f90938e47d99d45971937f4020d"
+    },
+    "metrics": {
+      "_dd.top_level": 1.0,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1.0,
+      "process_id": 91153.0
+    },
+    "duration": 2218000,
+    "start": 1780998997002290000
+  }],
+[
+  {
+    "name": "child-span",
+    "service": "fork-test",
+    "resource": "child-span",
+    "trace_id": 1,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "",
+    "error": 0,
+    "meta": {
+      "_dd.base_service": "ddtrace_subprocess_dir",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "6a27e35500000000",
+      "_dd.svc_src": "m",
+      "_dd.tags.process": "entrypoint.basedir:ddtrace_subprocess_dir,entrypoint.name:tmpuq52vj3g,entrypoint.type:script,entrypoint.workdir:dd-trace-py,svc.auto:ddtrace_subprocess_dir",
+      "language": "python",
+      "runtime-id": "53202310d1fd4f50a4eaa411ae87fa56"
+    },
+    "metrics": {
+      "_dd.top_level": 1.0,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1.0,
+      "process_id": 91156.0
+    },
+    "duration": 95000,
+    "start": 1780998997098421000
+  }]]