Fix gpu metrics (#56006)

cherrypick #56009

Bugs introduced in #52102, Two bugs:
- proc is a TypedDict so it needs to be fetched via `proc["pid"]`
instead of `proc.pid`.
- Changing `processes_pid` is backwards-incompatible change that ends up
changing the dashboard APIs that power the ray dashboard. Maintain
backwards-compatibility

Verified fix:
Metrics work again:
<img width="947" height="441" alt="Screenshot 2025-08-27 at 12 22 40 PM"
src="https://github.com/user-attachments/assets/0a9a83e7-b720-4ad0-b90e-1baa394edde5"
/>


Ray Dashboard works again:
<img width="1824" height="1029" alt="Screenshot 2025-08-27 at 12 21
51 PM"
src="https://github.com/user-attachments/assets/6b0e08e4-69c9-4223-b736-ff69b8d306db"
/>

---------

Signed-off-by: Alan Guo <[email protected]>

Author: alanwguo

python/ray/dashboard/modules/node/datacenter.py

@@ -225,7 +225,7 @@ async def _get_actor_info(actor: Optional[dict]) -> Optional[dict]:
                     break
 
             for gpu_stats in node_physical_stats.get("gpus", []):
-                # gpu_stats.get("processes") can be None, an empty list or a
+                # gpu_stats.get("processesPids") can be None, an empty list or a
                 # list of dictionaries.
                 for process in gpu_stats.get("processesPids") or []:
                     if process["pid"] == pid:

python/ray/dashboard/modules/reporter/gpu_providers.py

@@ -230,9 +230,9 @@ def _parse_nvsmi_pmon_output(
             1       7175     C     86     26      -      -      -      -    ray::TorchGPUWo
             2          -     -      -      -      -      -      -      -    -
 
-        Returns a dict mapping GPU index to list of ProcessGPUInfo.
+        Returns a dict mapping GPU index to dict of pid to ProcessGPUInfo.
         """
-        process_utilizations = defaultdict(list)
+        process_utilizations = defaultdict(dict)
         lines = nvsmi_stdout.splitlines()
         # Get the first line that is started with #
         table_header = None
@@ -275,7 +275,7 @@ def _parse_nvsmi_pmon_output(
                 ),  # Convert percentage to MB
                 gpu_utilization=sm,
             )
-            process_utilizations[gpu_id].append(process_info)
+            process_utilizations[gpu_id][pid] = process_info
         return process_utilizations
 
     def _get_pynvml_gpu_usage(self) -> List[GpuUtilizationInfo]:

python/ray/dashboard/modules/reporter/reporter_agent.py

@@ -882,7 +882,7 @@ def _get_agent_proc(self) -> psutil.Process:
     def _generate_worker_key(self, proc: psutil.Process) -> Tuple[int, float]:
         return (proc.pid, proc.create_time())
 
-    def _get_workers(self, gpus: Optional[List[GpuUtilizationInfo]] = None):
+    def _get_worker_processes(self):
         raylet_proc = self._get_raylet_proc()
         if raylet_proc is None:
             return []
@@ -899,7 +899,13 @@ def _get_workers(self, gpus: Optional[List[GpuUtilizationInfo]] = None):
                     self._generate_worker_key(proc): proc
                     for proc in raylet_proc.children()
                 }
+            return workers
 
+    def _get_workers(self, gpus: Optional[List[GpuUtilizationInfo]] = None):
+        workers = self._get_worker_processes()
+        if not workers:
+            return []
+        else:
             # We should keep `raylet_proc.children()` in `self` because
             # when `cpu_percent` is first called, it returns the meaningless 0.
             # See more: https://github.com/ray-project/ray/issues/29848
@@ -926,7 +932,7 @@ def _get_workers(self, gpus: Optional[List[GpuUtilizationInfo]] = None):
                     processes = gpu.get("processes_pids")
                     if processes:
                         for proc in processes.values():
-                            gpu_pid_mapping[proc.pid].append(proc)
+                            gpu_pid_mapping[proc["pid"]].append(proc)
 
             result = []
             for w in self._workers.values():
@@ -1752,6 +1758,15 @@ def _compose_stats_payload(
 
             self._metrics_agent.clean_all_dead_worker_metrics()
 
+        # Convert processes_pids back to a list of dictionaries to maintain backwards-compatibility
+        for gpu in stats["gpus"]:
+            if isinstance(gpu.get("processes_pids"), dict):
+                gpu["processes_pids"] = list(gpu["processes_pids"].values())
+
+        # TODO(aguo): Add a pydantic model for this dict to maintain compatibility
+        # with the Ray Dashboard API and UI code.
+
+        # NOTE: This converts keys to "Google style", (e.g: "processes_pids" -> "processesPids")
         return jsonify_asdict(stats)
 
     async def run(self, server):

python/ray/dashboard/modules/reporter/tests/test_reporter.py

@@ -1017,6 +1017,9 @@ def _get_agent_proc(self):
         def _generate_worker_key(self, proc):
             return (proc.pid, proc.create_time())
 
+        def _get_worker_processes(self):
+            return ReporterAgent._get_worker_processes(self)
+
     obj = ReporterAgentDummy()
 
     try: