Fix broken Otel propagation in dspy ParallelExecutor

dspy/utils/parallelizer.py

@@ -7,12 +7,37 @@
 import time
 import traceback
 from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
+from functools import wraps
 
 import tqdm
 
 logger = logging.getLogger(__name__)
 
 
+def _with_otel_context(otel_context):
+    """Decorator to attach OpenTelemetry context to a function."""
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            ctx_token = None
+            if otel_context:
+                try:
+                    from opentelemetry.context import attach, detach
+                    ctx_token = attach(otel_context)
+                except ImportError:
+                    pass
+            try:
+                return func(*args, **kwargs)
+            finally:
+                if ctx_token:
+                    try:
+                        detach(ctx_token)
+                    except:
+                        pass
+        return wrapper
+    return decorator
+
+
 class ParallelExecutor:
     def __init__(
         self,
@@ -75,7 +100,15 @@ def _execute_parallel(self, function, data):
         start_time_lock = threading.Lock()
         resubmitted = set()
 
+        # Capture OpenTelemetry context for trace preservation
+        try:
+            from opentelemetry.context import get_current
+            otel_context = get_current()
+        except ImportError:
+            otel_context = None
+
         # This is the worker function each thread will run.
+        @_with_otel_context(otel_context)
         def worker(parent_overrides, submission_id, index, item):
             if self.cancel_jobs.is_set():
                 return index, job_cancelled

pyproject.toml

@@ -66,7 +66,9 @@ test_extras = [
     "datasets>=2.14.6",
     "pandas>=2.1.1",
     "optuna>=3.4.0",
-    "langchain_core",
+    "langchain_core>=0.3.65",
+    "opentelemetry-api<3,>=1.9.0",
+    "opentelemetry-sdk<3,>=1.9.0"
 ]
 
 [tool.setuptools.packages.find]

tests/utils/test_parallelizer.py

@@ -1,3 +1,4 @@
+import importlib
 import time
 
 import pytest
@@ -59,3 +60,36 @@ def task(item):
 
     # Verify that the results exclude the failed task
     assert results == [1, 2, None, 4, 5]
+
+@pytest.mark.skipif(not importlib.util.find_spec("opentelemetry"), reason="OpenTelemetry not installed")
+@pytest.mark.extra
+def test_otel_context_propagation():
+    from opentelemetry import trace
+    from opentelemetry.sdk.trace import TracerProvider
+
+    tracer_provider = TracerProvider()
+    trace.set_tracer_provider(tracer_provider)
+    tracer = trace.get_tracer(__name__)
+
+    with tracer.start_as_current_span("parent_span") as parent_span:
+        parent_span.set_attribute("test_attribute", "parent_value")
+        parent_span_id = parent_span.get_span_context().span_id
+        parent_trace_id = parent_span.get_span_context().trace_id
+
+        seen_spans = []
+
+        def task(item):
+            with tracer.start_as_current_span("child_span") as child_span:
+                child_span.set_attribute("test_attribute", "child_value")
+                seen_spans.append(child_span)
+            return item * 2
+
+        executor = ParallelExecutor(num_threads=2, disable_progress_bar=True)
+        executor.execute(task, [1, 2, 3])
+
+        assert len(seen_spans) == 3
+
+        for span in seen_spans:
+            if span.name == "child_span":
+                assert span.parent.span_id == parent_span_id
+                assert span.get_span_context().trace_id == parent_trace_id