[Refactor][Core/Dashboard] Make datacenter.py a private implementation for NodeHead (#51847)

Signed-off-by: Chi-Sheng Liu <chishengliu@chishengliu.com>

Author: MortalHappiness

python/ray/dashboard/head.py

@@ -15,7 +15,6 @@
 from ray._raylet import GcsClient
 from ray.dashboard.consts import DASHBOARD_METRIC_PORT
 from ray.dashboard.dashboard_metrics import DashboardPrometheusMetrics
-from ray.dashboard.datacenter import DataOrganizer
 from ray.dashboard.utils import (
     DashboardHeadModule,
     DashboardHeadModuleConfig,
@@ -141,7 +140,6 @@ def __init__(
         self.gcs_error_subscriber = None
         self.gcs_log_subscriber = None
         self.ip = node_ip_address
-        DataOrganizer.head_node_ip = self.ip
 
         if self.minimal:
             self.server, self.grpc_port = None, None
@@ -445,8 +443,6 @@ async def _async_notify():
         concurrent_tasks = [
             self._gcs_check_alive(),
             _async_notify(),
-            DataOrganizer.purge(),
-            DataOrganizer.organize(self._executor),
         ]
         for m in dashboard_head_modules:
             concurrent_tasks.append(m.run(self.server))

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

@@ -39,8 +39,6 @@ class DataSource:
 
 
 class DataOrganizer:
-    head_node_ip = None
-
     @staticmethod
     @async_loop_forever(dashboard_consts.RAY_DASHBOARD_STATS_PURGING_INTERVAL)
     async def purge():

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

@@ -35,7 +35,7 @@
     DASHBOARD_AGENT_ADDR_NODE_ID_PREFIX,
     DASHBOARD_AGENT_ADDR_IP_PREFIX,
 )
-from ray.dashboard.datacenter import DataOrganizer, DataSource
+from ray.dashboard.modules.node.datacenter import DataOrganizer, DataSource
 from ray.dashboard.modules.node import node_consts
 from ray.dashboard.modules.node import actor_consts
 from ray.dashboard.utils import async_loop_forever
@@ -710,13 +710,40 @@ async def get_actor(self, req) -> aiohttp.web.Response:
             detail=actors[actor_id],
         )
 
+    @routes.get("/test/dump")
+    async def dump(self, req) -> aiohttp.web.Response:
+        """
+        Dump all data from datacenter. This is used for testing purpose only.
+        """
+        key = req.query.get("key")
+        if key is None:
+            all_data = {
+                k: dict(v)
+                for k, v in DataSource.__dict__.items()
+                if not k.startswith("_")
+            }
+            return dashboard_optional_utils.rest_response(
+                status_code=dashboard_utils.HTTPStatusCode.OK,
+                message="Fetch all data from datacenter success.",
+                **all_data,
+            )
+        else:
+            data = dict(DataSource.__dict__.get(key))
+            return dashboard_optional_utils.rest_response(
+                status_code=dashboard_utils.HTTPStatusCode.OK,
+                message=f"Fetch {key} from datacenter success.",
+                **{key: data},
+            )
+
     async def run(self, server):
         await asyncio.gather(
             self._update_nodes(),
             self._update_node_stats(),
             self._update_node_physical_stats(),
             self._update_actors(),
             self._cleanup_actors(),
+            DataOrganizer.purge(),
+            DataOrganizer.organize(self._node_executor),
         )
 
     @staticmethod

python/ray/dashboard/modules/tests/test_head.py

@@ -8,7 +8,6 @@
 import ray.dashboard.optional_utils as dashboard_optional_utils
 import ray.dashboard.utils as dashboard_utils
 from ray._private.ray_constants import env_bool
-from ray.dashboard.datacenter import DataSource
 
 logger = logging.getLogger(__name__)
 routes = dashboard_optional_utils.DashboardHeadRouteTable
@@ -41,28 +40,6 @@ async def route_delete(self, req) -> aiohttp.web.Response:
     async def route_view(self, req) -> aiohttp.web.Response:
         pass
 
-    @routes.get("/test/dump")
-    async def dump(self, req) -> aiohttp.web.Response:
-        key = req.query.get("key")
-        if key is None:
-            all_data = {
-                k: dict(v)
-                for k, v in DataSource.__dict__.items()
-                if not k.startswith("_")
-            }
-            return dashboard_optional_utils.rest_response(
-                status_code=dashboard_utils.HTTPStatusCode.OK,
-                message="Fetch all data from datacenter success.",
-                **all_data,
-            )
-        else:
-            data = dict(DataSource.__dict__.get(key))
-            return dashboard_optional_utils.rest_response(
-                status_code=dashboard_utils.HTTPStatusCode.OK,
-                message=f"Fetch {key} from datacenter success.",
-                **{key: data},
-            )
-
     @routes.get("/test/http_get")
     async def get_url(self, req) -> aiohttp.web.Response:
         url = req.query.get("url")