fix(futures) Allow calling ThreadPoolExecutor.submit with fn kwarg (#3035) (#3040)

(cherry picked from commit fb1c220)

Co-authored-by: Brett Langdon <[email protected]>

Author: mergify[bot]

ddtrace/contrib/futures/threading.py

@@ -21,10 +21,12 @@ def _wrap_submit(func, instance, args, kwargs):
     if ddtrace.tracer.context_provider._has_active_context():
         current_ctx = ddtrace.tracer.context_provider.active()
 
-    # extract the target function that must be executed in
-    # a new thread and the `target` arguments
-    fn = args[0]
-    fn_args = args[1:]
+    # The target function can be provided as a kwarg argument "fn" or the first positional argument
+    if "fn" in kwargs:
+        fn = kwargs.pop("fn")
+        fn_args = args
+    else:
+        fn, fn_args = args[0], args[1:]
     return func(_wrap_execution, current_ctx, fn, fn_args, kwargs)
 
 

releasenotes/notes/fix-futures-args-a206048e19889ad7.yaml

@@ -0,0 +1,3 @@
+fixes:
+  - |
+    Fix error when calling ``concurrent.futures.ThreadPoolExecutor.submit`` with ``fn`` keyword argument.

tests/contrib/futures/test_propagation.py

@@ -68,6 +68,28 @@ def fn(value, key=None):
             (dict(name="executor.thread"),),
         )
 
+    def test_propagation_with_kwargs(self):
+        # instrumentation must work if only kwargs are provided
+
+        def fn(value, key=None):
+            with self.tracer.trace("executor.thread"):
+                return value, key
+
+        with self.override_global_tracer():
+            with self.tracer.trace("main.thread"):
+                with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
+                    future = executor.submit(fn=fn, value=42, key="CheeseShop")
+                    value, key = future.result()
+                    # assert the right result
+                    self.assertEqual(value, 42)
+                    self.assertEqual(key, "CheeseShop")
+
+        # the trace must be completed
+        self.assert_structure(
+            dict(name="main.thread"),
+            (dict(name="executor.thread"),),
+        )
+
     def test_disabled_instrumentation(self):
         # it must not propagate if the module is disabled
         unpatch()