[Feat][Core/Dashboard] Convert ReportHead to subprocess module (#51733)

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

Author: MortalHappiness

python/ray/dashboard/head.py

@@ -297,6 +297,7 @@ def _load_subprocess_module_handles(
             cluster_id_hex=self.cluster_id_hex,
             gcs_address=self.gcs_address,
             session_name=self.session_name,
+            temp_dir=self.temp_dir,
             logging_level=self.logging_level,
             logging_format=self.logging_format,
             log_dir=self.log_dir,

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

@@ -26,11 +26,12 @@
 from ray.core.generated import reporter_pb2, reporter_pb2_grpc
 from ray.dashboard.consts import GCS_RPC_TIMEOUT_SECONDS
 from ray.dashboard.state_aggregator import StateAPIManager
+from ray.dashboard.subprocesses.routes import SubprocessRouteTable as routes
+from ray.dashboard.subprocesses.module import SubprocessModule
 from ray.util.state.common import ListApiOptions
 from ray.util.state.state_manager import StateDataSourceClient
 
 logger = logging.getLogger(__name__)
-routes = dashboard_optional_utils.DashboardHeadRouteTable
 
 EMOJI_WARNING = "⚠️"
 WARNING_FOR_MULTI_TASK_IN_A_WORKER = (
@@ -54,9 +55,9 @@
 )
 
 
-class ReportHead(dashboard_utils.DashboardHeadModule):
-    def __init__(self, config: dashboard_utils.DashboardHeadModuleConfig):
-        super().__init__(config)
+class ReportHead(SubprocessModule):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
         self._ray_config = None
         # TODO(fyrestone): Avoid using ray.state in dashboard, it's not
         # asynchronous and will lead to low performance. ray disconnect()
@@ -697,7 +698,8 @@ def _make_stub(
         channel = init_grpc_channel(ip_port, options=options, asynchronous=True)
         return reporter_pb2_grpc.ReporterServiceStub(channel)
 
-    async def run(self, server):
+    async def run(self):
+        await super().run()
         self._state_api_data_source_client = StateDataSourceClient(
             self.aiogrpc_gcs_channel, self.gcs_aio_client
         )
@@ -719,7 +721,3 @@ async def run(self, server):
             namespace=KV_NAMESPACE_CLUSTER,
         )
         self.cluster_metadata = json.loads(cluster_metadata.decode("utf-8"))
-
-    @staticmethod
-    def is_minimal_module():
-        return False

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

@@ -15,10 +15,12 @@
 import ray
 from ray._private import ray_constants
 from ray._private.metrics_agent import fix_grpc_metric
+import ray._private.usage.usage_lib as ray_usage_lib
 from ray._private.test_utils import (
     fetch_prometheus,
     format_web_url,
     wait_for_condition,
+    wait_until_server_available,
 )
 from ray.core.generated.metrics_pb2 import Metric
 from ray.dashboard.modules.reporter.reporter_agent import ReporterAgent
@@ -948,5 +950,20 @@ def verify():
     wait_for_condition(verify, timeout=10)
 
 
+def test_get_cluster_metadata(ray_start_with_dashboard):
+    assert wait_until_server_available(ray_start_with_dashboard["webui_url"])
+    webui_url = format_web_url(ray_start_with_dashboard["webui_url"])
+    url = f"{webui_url}/api/v0/cluster_metadata"
+
+    resp = requests.get(url)
+    assert resp.status_code == 200
+    resp_data = resp.json()["data"]
+    meta = ray_usage_lib._generate_cluster_metadata(ray_init_cluster=True)
+    assert len(resp_data) == len(meta)
+    assert resp_data["pythonVersion"] == meta["python_version"]
+    assert resp_data["rayVersion"] == meta["ray_version"]
+    assert resp_data["rayInitCluster"] == meta["ray_init_cluster"]
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main(["-v", __file__]))

python/ray/dashboard/subprocesses/module.py

@@ -35,6 +35,7 @@ class SubprocessModuleConfig:
     cluster_id_hex: str
     gcs_address: str
     session_name: str
+    temp_dir: str
     # Logger configs. Will be set up in subprocess entrypoint `run_module`.
     logging_level: str
     logging_format: str
@@ -175,6 +176,10 @@ def session_name(self):
         """
         return self._config.session_name
 
+    @property
+    def temp_dir(self):
+        return self._config.temp_dir
+
     @property
     def log_dir(self):
         return self._config.log_dir

python/ray/dashboard/subprocesses/tests/test_e2e.py

@@ -27,6 +27,7 @@ def default_module_config(tmp_path) -> SubprocessModuleConfig:
         cluster_id_hex="test_cluster_id",
         gcs_address="",
         session_name="test_session",
+        temp_dir=str(tmp_path),
         logging_level=ray_constants.LOGGER_LEVEL,
         logging_format=ray_constants.LOGGER_FORMAT,
         log_dir=str(tmp_path),