chore(ray): add instrumentation for ray.get (#14768)

## Description

This PR adds instrumentation for ray.get to our Ray integration, as
specified in
[MLOB-3820](https://datadoghq.atlassian.net/browse/MLOB-3820). This is
based on #14587, and is a
clean recreation of #14658
with some minor changes.

## Testing

**NOTE (10/3 12PM): All tests re-run after latest commit**

I launched the testagent with docker compose up -d testagent, and then
in a scripts/ddtest container I ran `DD_AGENT_PORT=9126 riot -v run
--pass-env -p 3.12 ray -- -s -vv` to execute the snapshot tests, which
resulted in all tests passing. I committed and pushed the generated
snapshot tests jsons to this branch.

<img width="1131" height="171" alt="Screenshot 2025-10-03 at 11 28
59 AM"
src="https://github.com/user-attachments/assets/791f328e-9adf-463b-8a79-b822426c0ca6"
/>

Manual test on local Ray cluster:

Create/activate ray venv and pip install dd-trace-py from local repo
with these changes.

```
export DD_SERVICE="imran-ray-get-test-002"
export DD_TRACE_RAY_CORE_API=1
RAY_LOGGING_CONFIG_ENCODING=JSON DD_ENV=dev ray start --head --dashboard-host=127.0.0.1 --tracing-startup-hook=ddtrace.contrib.ray:setup_tracing
```
submit the
[simple_get.py](https://github.com/user-attachments/files/22682834/simple_get.py)
test job to local cluster with:

`ray job submit --submission-id="imran-ray-get-test-002" -- python
/Users/imran.hendley/scripts/simple_get.py`

And we can see the ray.get span on the job:

<img width="1217" height="933" alt="Screenshot 2025-10-03 at 11 49
28 AM"
src="https://github.com/user-attachments/assets/6feb4530-d1f1-43bb-83aa-99d3d424c315"
/>


The same span in trace explorer is
[here](https://dd.datad0g.com/apm/trace/68dff038000000001c792f1f5460456c?graphType=json&shouldShowLegend=true&spanID=18012011505180865563&timeHint=1759506488231.9692&traceQuery=).

## Risks

None

## Additional Notes

None

[MLOB-3820]:
https://datadoghq.atlassian.net/browse/MLOB-3820?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Author: imran-at-datadog

ddtrace/contrib/internal/ray/constants.py

@@ -40,6 +40,7 @@
 RAY_WAIT_TIMEOUT = "ray.wait.timeout_s"
 RAY_WAIT_NUM_RETURNS = "ray.wait.num_returns"
 RAY_WAIT_FETCH_LOCAL = "ray.wait.fetch_local"
+RAY_GET_VALUE_SIZE_BYTES = "ray.get.value_size_bytes"
 RAY_PUT_VALUE_TYPE = "ray.put.value_type"
 RAY_PUT_VALUE_SIZE_BYTES = "ray.put.value_size_bytes"
 RAY_METADATA_PREFIX = "ray.job.metadata"

ddtrace/contrib/internal/ray/patch.py

@@ -29,6 +29,7 @@
 from .constants import RAY_ACTOR_METHOD_ARGS
 from .constants import RAY_ACTOR_METHOD_KWARGS
 from .constants import RAY_ENTRYPOINT
+from .constants import RAY_GET_VALUE_SIZE_BYTES
 from .constants import RAY_JOB_NAME
 from .constants import RAY_JOB_STATUS
 from .constants import RAY_JOB_SUBMIT_STATUS
@@ -287,6 +288,33 @@ def traced_actor_method_call(wrapped, instance, args, kwargs):
         return wrapped(*args, **kwargs)
 
 
+def traced_get(wrapped, instance, args, kwargs):
+    """
+    Trace the calls of ray.get
+    """
+    if not config.ray.trace_core_api:
+        return wrapped(*args, **kwargs)
+
+    if tracer.current_span() is None:
+        tracer.context_provider.activate(_extract_tracing_context_from_env())
+
+    with long_running_ray_span(
+        "ray.get",
+        service=RAY_SERVICE_NAME or DEFAULT_JOB_NAME,
+        span_type=SpanTypes.RAY,
+        child_of=tracer.context_provider.active(),
+        activate=True,
+    ) as span:
+        span.set_tag_str(SPAN_KIND, SpanKind.PRODUCER)
+        timeout = kwargs.get("timeout")
+        if timeout is not None:
+            span.set_tag_str("ray.get.timeout_s", str(timeout))
+        _inject_ray_span_tags_and_metrics(span)
+        get_value = get_argument_value(args, kwargs, 0, "object_refs")
+        span.set_tag_str(RAY_GET_VALUE_SIZE_BYTES, str(sys.getsizeof(get_value)))
+        return wrapped(*args, **kwargs)
+
+
 def traced_put(wrapped, instance, args, kwargs):
     """
     Trace the calls of ray.put
@@ -529,6 +557,7 @@ def _(m):
     def _(m):
         _w(m.RemoteFunction, "_remote", traced_submit_task)
 
+    _w(ray, "get", traced_get)
     _w(ray, "wait", traced_wait)
     _w(ray, "put", traced_put)
 
@@ -545,6 +574,7 @@ def unpatch():
     _u(ray.actor, "_modify_class")
     _u(ray.actor.ActorHandle, "_actor_method_call")
 
+    _u(ray, "get")
     _u(ray, "wait")
     _u(ray, "put")
 

tests/contrib/ray/test_ray.py

@@ -215,6 +215,17 @@ def add_one(x):
         assert running == [], f"Expected no running tasks, got {len(running)}"
         assert ray.get(done) == [43], f"Expected done to be [43], got {done}"
 
+    @pytest.mark.snapshot(token="tests.contrib.ray.test_ray.test_simple_get", ignores=RAY_SNAPSHOT_IGNORES)
+    def test_simple_get(self):
+        with override_config("ray", dict(trace_core_api=True)):
+
+            @ray.remote
+            def add_one(x):
+                return x + 1
+
+            results = ray.get([add_one.remote(x) for x in range(1)])
+            assert results == [1], f"Expected [1], got {results}"
+
     @pytest.mark.snapshot(token="tests.contrib.ray.test_ray.test_simple_wait", ignores=RAY_SNAPSHOT_IGNORES)
     def test_simple_wait(self):
         with override_config("ray", dict(trace_core_api=True)):

tests/contrib/ray/test_ray_patch.py

@@ -21,6 +21,7 @@ def assert_module_patched(self, ray):
         self.assert_wrapped(ray.dashboard.modules.job.job_manager.JobManager._monitor_job_internal)
         self.assert_wrapped(ray.actor._modify_class)
         self.assert_wrapped(ray.actor.ActorHandle._actor_method_call)
+        self.assert_wrapped(ray.get)
         self.assert_wrapped(ray.wait)
         self.assert_wrapped(ray.put)
 
@@ -30,14 +31,16 @@ def assert_not_module_patched(self, ray):
         self.assert_not_wrapped(ray.dashboard.modules.job.job_manager.JobManager._monitor_job_internal)
         self.assert_not_wrapped(ray.actor._modify_class)
         self.assert_not_wrapped(ray.actor.ActorHandle._actor_method_call)
-        self.assert_not_wrapped(ray.put)
+        self.assert_not_wrapped(ray.get)
         self.assert_not_wrapped(ray.wait)
+        self.assert_not_wrapped(ray.put)
 
     def assert_not_module_double_patched(self, ray):
         self.assert_not_double_wrapped(ray.remote_function.RemoteFunction._remote)
         self.assert_not_double_wrapped(ray.dashboard.modules.job.job_manager.JobManager.submit_job)
         self.assert_not_double_wrapped(ray.dashboard.modules.job.job_manager.JobManager._monitor_job_internal)
         self.assert_not_double_wrapped(ray.actor._modify_class)
         self.assert_not_double_wrapped(ray.actor.ActorHandle._actor_method_call)
+        self.assert_not_double_wrapped(ray.get)
         self.assert_not_double_wrapped(ray.wait)
         self.assert_not_double_wrapped(ray.put)

tests/snapshots/tests.contrib.ray.test_ray.test_simple_get.json

@@ -0,0 +1,103 @@
+[[
+  {
+    "name": "task.submit",
+    "service": "tests.contrib.ray",
+    "resource": "tests.contrib.ray.test_ray.add_one.remote",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "ray",
+    "error": 0,
+    "meta": {
+      "_dd.hostname": "docker-desktop",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "68dfeb8100000000",
+      "component": "ray",
+      "language": "python",
+      "ray.hostname": "docker-desktop",
+      "ray.job_id": "01000000",
+      "ray.node_id": "142aa12894d2b47a2b01989991cd0cec0340e675b8ba6d22a3ea0292",
+      "ray.task.submit_status": "success",
+      "ray.worker_id": "01000000ffffffffffffffffffffffffffffffffffffffffffffffff",
+      "runtime-id": "81dd04c9904045e9b283e0bdc8fe152f",
+      "span.kind": "producer"
+    },
+    "metrics": {
+      "_dd.djm.enabled": 1,
+      "_dd.filter.kept": 1,
+      "_dd.measured": 1,
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 2,
+      "process_id": 1289
+    },
+    "duration": 2621500,
+    "start": 1759505281704904470
+  },
+     {
+       "name": "task.execute",
+       "service": "tests.contrib.ray",
+       "resource": "tests.contrib.ray.test_ray.TestRayIntegration.test_simple_get.<locals>.add_one",
+       "trace_id": 0,
+       "span_id": 2,
+       "parent_id": 1,
+       "type": "ray",
+       "error": 0,
+       "meta": {
+         "component": "ray",
+         "ray.hostname": "docker-desktop",
+         "ray.job_id": "01000000",
+         "ray.node_id": "142aa12894d2b47a2b01989991cd0cec0340e675b8ba6d22a3ea0292",
+         "ray.task.status": "success",
+         "ray.worker_id": "01000000ffffffffffffffffffffffffffffffffffffffffffffffff",
+         "runtime-id": "81dd04c9904045e9b283e0bdc8fe152f",
+         "span.kind": "consumer"
+       },
+       "metrics": {
+         "_dd.djm.enabled": 1,
+         "_dd.filter.kept": 1,
+         "_dd.measured": 1,
+         "_dd.top_level": 1,
+         "_sampling_priority_v1": 2,
+         "process_id": 1289
+       },
+       "duration": 266417,
+       "start": 1759505281707023595
+     }],
+[
+  {
+    "name": "ray.get",
+    "service": "unnamed.ray.job",
+    "resource": "ray.get",
+    "trace_id": 1,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "ray",
+    "error": 0,
+    "meta": {
+      "_dd.base_service": "tests.contrib.ray",
+      "_dd.hostname": "docker-desktop",
+      "_dd.p.dm": "-0",
+      "_dd.p.tid": "68dfeb8100000000",
+      "component": "ray",
+      "language": "python",
+      "ray.get.value_size_bytes": "88",
+      "ray.hostname": "docker-desktop",
+      "ray.job_id": "01000000",
+      "ray.node_id": "142aa12894d2b47a2b01989991cd0cec0340e675b8ba6d22a3ea0292",
+      "ray.worker_id": "01000000ffffffffffffffffffffffffffffffffffffffffffffffff",
+      "runtime-id": "81dd04c9904045e9b283e0bdc8fe152f",
+      "span.kind": "producer"
+    },
+    "metrics": {
+      "_dd.djm.enabled": 1,
+      "_dd.filter.kept": 1,
+      "_dd.measured": 1,
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 2,
+      "process_id": 1289
+    },
+    "duration": 313000,
+    "start": 1759505281707677970
+  }]]