Make the DatadogExportSpanProcessor re-initialize itself after fork

Author: phillipuniverse

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#903](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/903))
 - `opentelemetry-instrumentation-falcon` Safer patching mechanism
   ([#895](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/895))
+- `opentelemetry-exporter-datadog` re-initializes itself on fork to prevent deadlocks ([#932](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/932))
 
 ## [1.9.1-0.28b1](https://github.com/open-telemetry/opentelemetry-python/releases/tag/v1.9.1-0.28b1) - 2022-01-29
 

exporter/opentelemetry-exporter-datadog/src/opentelemetry/exporter/datadog/spanprocessor.py

@@ -14,6 +14,7 @@
 
 import collections
 import logging
+import os
 import threading
 import typing
 
@@ -67,7 +68,7 @@ def __init__(
         self.traces_spans_count = collections.Counter()
         self.traces_spans_ended_count = collections.Counter()
 
-        self.worker_thread = threading.Thread(target=self.worker, daemon=True)
+        self.worker_thread = self._create_worker_thread()
 
         # threading conditions used for flushing and shutdown
         self.condition = threading.Condition(threading.Lock())
@@ -81,6 +82,24 @@ def __init__(
         self.schedule_delay_millis = schedule_delay_millis
         self.done = False
         self.worker_thread.start()
+        # Only available in *nix since py37.
+        if hasattr(os, "register_at_fork"):
+            os.register_at_fork(
+                after_in_child=self._at_fork_reinit
+            )  # pylint: disable=protected-access
+
+    def _at_fork_reinit(self):
+        self.condition = threading.Condition(threading.Lock())
+        self.flush_condition = threading.Condition(threading.Lock())
+        self.traces_lock = threading.Lock()
+        self.check_traces_queue.clear()
+        self.worker_thread = self._create_worker_thread()
+        self.worker_thread.start()
+
+    def _create_worker_thread(self) -> threading.Thread:
+        return threading.Thread(
+            name="DatadogProcessor", target=self.worker, daemon=True
+        )
 
     def on_start(
         self, span: Span, parent_context: typing.Optional[Context] = None

exporter/opentelemetry-exporter-datadog/tests/test_datadog_exporter.py

@@ -14,6 +14,8 @@
 
 import itertools
 import logging
+import multiprocessing
+import os
 import sys
 import time
 import unittest
@@ -30,6 +32,7 @@
 from opentelemetry.sdk.trace import Resource, sampling
 from opentelemetry.sdk.util.instrumentation import InstrumentationInfo
 from opentelemetry.semconv.trace import SpanAttributes
+from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 
 
 class MockDatadogSpanExporter(datadog.DatadogSpanExporter):
@@ -41,6 +44,9 @@ def __init__(self, *args, **kwargs):
         agent_writer_mock.exit_timeout = 1
         self._agent_writer = agent_writer_mock
 
+    def reset(self):
+        self._agent_writer.reset_mock()
+
 
 def get_spans(tracer, exporter, shutdown=True):
     if shutdown:
@@ -667,3 +673,53 @@ def test_service_name_fallback(self):
 
         span = datadog_spans[0]
         self.assertEqual(span["service"], "fallback_service_name")
+
+
+class TestDatadogSpanProcessorConcurrency(ConcurrencyTestBase):
+
+    @unittest.skipUnless(
+        hasattr(os, "fork") and sys.version_info >= (3, 7),
+        "needs *nix and minor version 7 or later",
+    )
+    def test_exports_with_fork(self):
+        # pylint: disable=invalid-name
+        tracer_provider = trace.TracerProvider()
+        tracer = tracer_provider.get_tracer(__name__)
+
+        exporter = MockDatadogSpanExporter()
+
+        span_processor = datadog.DatadogExportSpanProcessor(
+            exporter, schedule_delay_millis=50
+        )
+        tracer_provider.add_span_processor(span_processor)
+        with tracer.start_as_current_span("foo"):
+            pass
+        time.sleep(0.5)  # give some time for the exporter to upload spans
+
+        self.assertTrue(span_processor.force_flush())
+        self.assertEqual(len(get_spans(tracer, exporter, shutdown=False)), 1)
+        exporter.reset()
+
+        def child(conn):
+            def _target():
+                with tracer.start_as_current_span("span") as s:
+                    s.set_attribute("i", "1")
+                    with tracer.start_as_current_span("temp"):
+                        pass
+
+            self.run_with_many_threads(_target, 100)
+
+            time.sleep(0.5)
+
+            spans = get_spans(tracer, exporter, shutdown=False)
+            conn.send(len(spans) == 200)
+            conn.close()
+
+        parent_conn, child_conn = multiprocessing.Pipe()
+        fork_context = multiprocessing.get_context("fork")
+        p = fork_context.Process(target=child, args=(child_conn,))
+        p.start()
+        self.assertTrue(parent_conn.recv())
+        p.join()
+
+        span_processor.shutdown()