[Core] Cherry pick #47714, #47701 (#47724)

pick #47714, #47701

python/ray/dashboard/datacenter.py

@@ -87,8 +87,7 @@ async def organize(cls, thread_pool_executor):
                 node_stats,
             )
             for worker in workers:
-                stats = worker.get("coreWorkerStats", {})
-                if stats:
+                for stats in worker.get("coreWorkerStats", []):
                     worker_id = stats["workerId"]
                     core_worker_stats[worker_id] = stats
             node_workers[node_id] = workers
@@ -113,7 +112,9 @@ def merge_workers_for_node(cls, node_physical_stats, node_stats):
         for worker in node_physical_stats.get("workers", []):
             worker = dict(worker)
             pid = worker["pid"]
-            worker["coreWorkerStats"] = pid_to_worker_stats.get(pid, {})
+            core_worker_stats = pid_to_worker_stats.get(pid)
+            # Empty list means core worker stats is not available.
+            worker["coreWorkerStats"] = [core_worker_stats] if core_worker_stats else []
             worker["language"] = pid_to_language.get(
                 pid, dashboard_consts.DEFAULT_LANGUAGE
             )
@@ -147,8 +148,6 @@ async def get_node_info(cls, node_id, get_summary=False):
         node_info["raylet"] = node_stats
         node_info["raylet"].update(ray_stats)
 
-        node_info["status"] = node["stateSnapshot"]["state"]
-
         # Merge GcsNodeInfo to node physical stats
         node_info["raylet"].update(node)
         death_info = node.get("deathInfo", {})