feat(ray): add ignored actors configuration (#17673)

This PR adds the possibility to ignore actors. Ray by default will instrument all actors which can be super noisy. It is not possible to have the full list of ray actors we want to ignore while creating the integration.

Therefore, we give a config for any user to add any actor to not instrument

Co-authored-by: louis.tricot <louis.tricot@datadoghq.com>

Author: dubloom

ddtrace/contrib/internal/ray/__init__.py

@@ -40,6 +40,11 @@
     Enable it when you need visibility into queuing/submission latency, fan-out patterns,
     or when work is submitted but delayed before execution.
 
+- ``DD_TRACE_RAY_IGNORED_ACTORS``: JSON object mapping actor class names to actor methods
+    to exclude from instrumentation, or ``"*"`` to exclude the full actor class (default:
+    empty). For example, ``{"ActorA": ["method1"], "ActorB": "*"}``. Matching is based
+    on class name only, not module.
+
 - ``DD_TRACE_EXPERIMENTAL_LONG_RUNNING_FLUSH_INTERVAL``: Interval for resubmitting long-running
     spans (default: ``120.0`` seconds)
 

ddtrace/contrib/internal/ray/core/actor.py

@@ -208,6 +208,7 @@ def inject_tracing_into_actor_class(wrapped, instance, args, kwargs):
     cls = wrapped(*args, **kwargs)
     module_name = str(cls.__module__)
     class_name = str(cls.__name__)
+    ignored_actor_methods = _get_ray_integration_config().ignored_actors.get(class_name, frozenset())
 
     # Skip tracing for certain ray modules
     if any(module_name.startswith(denied_module) for denied_module in RAY_ACTOR_MODULE_DENYLIST):
@@ -217,14 +218,18 @@ def inject_tracing_into_actor_class(wrapped, instance, args, kwargs):
     if class_name.startswith("_"):
         return cls
 
+    # Allow users to skip instrumentation for specific actor class names.
+    if "*" in ignored_actor_methods:
+        return cls
+
     # Determine if the class is a JobSupervisor
     is_job_supervisor = f"{module_name}.{class_name}" == "ray.dashboard.modules.job.job_supervisor.JobSupervisor"
     # We do not want to instrument ping and polling to remove noise
     methods_to_ignore = {"ping", "_polling"} if is_job_supervisor else set()
 
     methods = inspect.getmembers(cls, is_function_or_method)
     for name, method in methods:
-        if name in methods_to_ignore:
+        if name in methods_to_ignore or name in ignored_actor_methods:
             continue
 
         if (

ddtrace/contrib/internal/ray/patch.py

@@ -1,3 +1,7 @@
+from collections.abc import Mapping
+import json
+from typing import Any
+
 import ray
 from wrapt import wrap_function_wrapper as _w
 
@@ -31,6 +35,45 @@
 _job_contexts = {}
 
 
+def _parse_ignored_actors(value: Any) -> dict[str, frozenset[str]]:
+    if not value:
+        return {}
+
+    if isinstance(value, str):
+        try:
+            parsed_value = json.loads(value)
+        except ValueError:
+            log.warning("Invalid DD_TRACE_RAY_IGNORED_ACTORS value. Expected a JSON object.")
+            return {}
+    else:
+        parsed_value = value
+
+    if not isinstance(parsed_value, Mapping):
+        log.warning("Invalid DD_TRACE_RAY_IGNORED_ACTORS value. Expected a JSON object.")
+        return {}
+
+    ignored_actors = {}
+    for actor_name, ignored_methods in parsed_value.items():
+        if not isinstance(actor_name, str) or not actor_name.strip():
+            log.warning("Invalid DD_TRACE_RAY_IGNORED_ACTORS actor name. Expected a non-empty string.")
+            continue
+
+        actor_name = actor_name.strip()
+        if ignored_methods == "*":
+            ignored_actors[actor_name] = frozenset({"*"})
+            continue
+
+        if not isinstance(ignored_methods, (list, tuple, set, frozenset)):
+            log.warning("Invalid DD_TRACE_RAY_IGNORED_ACTORS methods for actor %s. Expected a list.", actor_name)
+            continue
+
+        methods = frozenset(method.strip() for method in ignored_methods if isinstance(method, str) and method.strip())
+        if methods:
+            ignored_actors[actor_name] = methods
+
+    return ignored_actors
+
+
 config._add(
     "ray",
     dict(
@@ -39,6 +82,7 @@
         trace_core_api=_get_config("DD_TRACE_RAY_CORE_API", default=False, modifier=asbool),
         trace_args_kwargs=_get_config("DD_TRACE_RAY_ARGS_KWARGS", default=False, modifier=asbool),
         submission_spans=_get_config("DD_TRACE_RAY_SUBMISSION_SPANS_ENABLED", default=False, modifier=asbool),
+        ignored_actors=_get_config("DD_TRACE_RAY_IGNORED_ACTORS", default={}, modifier=_parse_ignored_actors),
     ),
 )
 

releasenotes/notes/feat-ray-instrument-actor-config-e072a42b63425d28.yaml

@@ -0,0 +1,6 @@
+---
+upgrade:
+  - |
+    ray: Adds ``DD_TRACE_RAY_IGNORED_ACTORS`` configuration to exclude specific Ray actor methods from instrumentation.
+    Set ``DD_TRACE_RAY_IGNORED_ACTORS='{"ActorA": ["method1"], "ActorB": "*"}'`` to leave matching methods or actors uninstrumented while continuing to trace other Ray actor methods.
+    Matching is based on actor class name only.

tests/contrib/ray/test_ray.py

@@ -7,6 +7,7 @@
 import ray
 from ray.util.tracing import tracing_helper
 
+from ddtrace.contrib.internal.ray.constants import DD_RAY_TRACE_CTX
 from tests.utils import TracerTestCase
 from tests.utils import override_config
 
@@ -39,6 +40,16 @@
 ]
 
 
+def test_parse_ignored_actors_returns_actor_method_mapping():
+    from ddtrace.contrib.internal.ray.patch import _parse_ignored_actors
+
+    assert _parse_ignored_actors("") == {}
+    assert _parse_ignored_actors('{"IgnoredCounter": ["increment"], "IgnoredActor": "*"}') == {
+        "IgnoredCounter": frozenset({"increment"}),
+        "IgnoredActor": frozenset({"*"}),
+    }
+
+
 class TestRayIntegration(TracerTestCase):
     """Test Ray integration with actual cluster setup and job submission"""
 
@@ -129,6 +140,46 @@ def get_value(self):
         value = ray.get(denied_actor.get_value.remote())
         assert value == 42, f"Unexpected result: {value}"
 
+    def test_configurable_ignored_actor_methods(self):
+        with override_config(
+            "ray",
+            dict(
+                ignored_actors={"PartiallyIgnoredCounter": frozenset({"increment"}), "IgnoredCounter": frozenset({"*"})}
+            ),
+        ):
+
+            @ray.remote
+            class PartiallyIgnoredCounter:
+                def increment(self):
+                    return 1
+
+                def get_value(self):
+                    return 1
+
+            @ray.remote
+            class IgnoredCounter:
+                def increment(self):
+                    return 1
+
+            partially_ignored_actor = PartiallyIgnoredCounter.remote()
+            ignored_actor = IgnoredCounter.remote()
+
+            partially_ignored_result = ray.get(partially_ignored_actor.increment.remote())
+            traced_result = ray.get(partially_ignored_actor.get_value.remote())
+            ignored_result = ray.get(ignored_actor.increment.remote())
+
+            assert partially_ignored_result == 1
+            assert ignored_result == 1
+            assert traced_result == 1
+
+            partially_ignored_params = [p.name for p in partially_ignored_actor._ray_method_signatures["increment"]]
+            traced_params = [p.name for p in partially_ignored_actor._ray_method_signatures["get_value"]]
+            ignored_params = [p.name for p in ignored_actor._ray_method_signatures["increment"]]
+
+            assert DD_RAY_TRACE_CTX not in partially_ignored_params
+            assert DD_RAY_TRACE_CTX not in ignored_params
+            assert DD_RAY_TRACE_CTX in traced_params
+
     @pytest.mark.snapshot(token="tests.contrib.ray.test_ray.test_ignored_tasks", ignores=RAY_SNAPSHOT_IGNORES)
     def test_ignored_tasks(self):
         """Test that tasks from modules in RAY_TASK_MODULE_DENYLIST are not traced."""