feat(ray): better handle the absence of ray.init()

Author: dubloom

ddtrace/contrib/internal/ray/__init__.py

@@ -11,16 +11,19 @@
 you have to follow one of the two methods below:
 
 The recommended way to instrument Ray, is to instrument the Ray cluster.
-You can do it by starting the Ray head with a tracing startup hook::
 
-    ray start --head --tracing-startup-hook=ddtrace.contrib.ray:setup_tracing
+You can do it by starting the Ray head using ddtrace-run::
 
-Otherwise, you can specify the tracing hook in `ray.init()` using::
+    DD_PATCH_MODULES="ray:true, aiohttp:false, grpc:false, requests:false" ddtrace-run ray start --head
 
-    ray.init(_tracing_startup_hook="ddtrace.contrib.ray:setup_tracing")
+DD_PATCH_MODULES will allow to reduce noise.
 
-Note that this method does not provide full tracing capabilities.
+You can do it by starting the Ray head with a tracing startup hook::
+
+    ray start --head --tracing-startup-hook=ddtrace.contrib.ray:setup_tracing
 
+Note that this method does not provide full tracing capabilities if ``ray.init()`` is not called at the top
+of your job scripts.
 
 Configuration
 ~~~~~~~~~~~~~

ddtrace/contrib/internal/ray/patch.py

@@ -153,11 +153,10 @@ def traced_submit_task(wrapped, instance, args, kwargs):
     # This is done under a lock as multiple task could be submit at the same time
     # and thus try to modify the signature as the same time
     with instance._inject_lock:
-        if not getattr(instance._function, "_dd_trace_wrapped", False):
+        if instance._function_signature is None:
             instance._function = _wrap_remote_function_execution(instance._function)
             instance._function.__signature__ = _inject_dd_trace_ctx_kwarg(instance._function)
             instance._function_signature = extract_signature(instance._function)
-            instance._function._dd_trace_wrapped = True
 
     with tracer.trace(
         "task.submit",

tests/contrib/ray/jobs/actor_and_task.py

@@ -0,0 +1,24 @@
+import ray
+
+
+@ray.remote
+class Counter:
+    def __init__(self):
+        self.value = 0
+
+    def increment(self):
+        self.value += 1
+        return self.value
+
+    def get_value(self):
+        return self.value
+
+
+def main():
+    counter = Counter.remote()
+    result = ray.get(counter.increment.remote())
+    assert result == 1, f"Expected 1, got {result}"
+
+
+if __name__ == "__main__":
+    main()