[Feat][Core/Dashboard] Convert EventHead to subprocess module (#51587)

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

Author: MortalHappiness

python/ray/dashboard/modules/event/event_head.py

@@ -10,6 +10,7 @@
 
 import aiohttp.web
 
+import ray
 import ray.dashboard.optional_utils as dashboard_optional_utils
 import ray.dashboard.utils as dashboard_utils
 from ray._common.utils import get_or_create_event_loop
@@ -22,10 +23,11 @@
 )
 from ray.dashboard.modules.event.event_utils import monitor_events, parse_event_strings
 from ray.dashboard.state_api_utils import do_filter, handle_list_api
+from ray.dashboard.subprocesses.routes import SubprocessRouteTable as routes
+from ray.dashboard.subprocesses.module import SubprocessModule
 from ray.util.state.common import ClusterEventState, ListApiOptions, ListApiResponse
 
 logger = logging.getLogger(__name__)
-routes = dashboard_optional_utils.DashboardHeadRouteTable
 
 JobEvents = OrderedDict
 dashboard_utils._json_compatible_types.add(JobEvents)
@@ -79,11 +81,11 @@ def transform(all_events) -> ListApiResponse:
 
 
 class EventHead(
-    dashboard_utils.DashboardHeadModule,
+    SubprocessModule,
     dashboard_utils.RateLimitedModule,
 ):
-    def __init__(self, config: dashboard_utils.DashboardHeadModuleConfig):
-        dashboard_utils.DashboardHeadModule.__init__(self, config)
+    def __init__(self, *args, **kwargs):
+        SubprocessModule.__init__(self, *args, **kwargs)
         dashboard_utils.RateLimitedModule.__init__(
             self,
             min(
@@ -105,6 +107,10 @@ def __init__(self, config: dashboard_utils.DashboardHeadModuleConfig):
             thread_name_prefix="event_head_executor",
         )
 
+        # To init gcs_client in internal_kv for record_extra_usage_tag.
+        assert self.gcs_client is not None
+        assert ray.experimental.internal_kv._internal_kv_initialized()
+
     async def limit_handler_(self):
         return dashboard_optional_utils.rest_response(
             status_code=dashboard_utils.HTTPStatusCode.INTERNAL_ERROR,
@@ -148,8 +154,15 @@ async def report_events(self, request):
         The request body is a JSON array of event strings in type string.
         Response should contain {"success": true}.
         """
-        request_body: List[str] = await request.json()
-        events = [parse_event_strings(event_str) for event_str in request_body]
+        try:
+            request_body: List[str] = await request.json()
+        except Exception as e:
+            logger.warning(f"Failed to parse request body: {request=}, {e=}")
+            raise aiohttp.web.HTTPBadRequest()
+        if not isinstance(request_body, list):
+            logger.warning(f"Request body is not a list, {request_body=}")
+            raise aiohttp.web.HTTPBadRequest()
+        events = parse_event_strings(request_body)
         logger.debug("Received %d events", len(events))
         self._update_events(events)
         self.total_report_events_count += 1
@@ -208,13 +221,10 @@ async def list_api_fn(option: ListApiOptions):
 
         return await handle_list_api(list_api_fn, req)
 
-    async def run(self, server):
+    async def run(self):
+        await super().run()
         self._monitor = monitor_events(
             self._event_dir,
             lambda data: self._update_events(parse_event_strings(data)),
             self._executor,
         )
-
-    @staticmethod
-    def is_minimal_module():
-        return False

python/ray/dashboard/modules/event/tests/test_event.py

@@ -231,6 +231,31 @@ def _check_events():
     wait_for_condition(_check_events, timeout=15)
 
 
+def test_report_events(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}/report_events"
+
+    resp = requests.post(url)
+    assert resp.status_code == 400
+    resp = requests.post(url, json={"Hello": "World"})
+    assert resp.status_code == 400
+
+    job_id = ray.JobID.from_int(100).hex()
+    sample_event = _get_event("Hello", job_id=job_id)
+    resp = requests.post(url, json=[json.dumps(sample_event)])
+    assert resp.status_code == 200
+
+    resp = requests.get(f"{webui_url}/events")
+    assert resp.status_code == 200
+    result = resp.json()
+    all_events = result["data"]["events"]
+    assert len(all_events) == 1
+    assert job_id in all_events
+    assert len(all_events[job_id]) == 1
+    assert all_events[job_id][0]["message"] == "Hello"
+
+
 @pytest.mark.asyncio
 async def test_monitor_events():
     with tempfile.TemporaryDirectory() as temp_dir:

python/ray/dashboard/subprocesses/module.py

@@ -163,6 +163,10 @@ def session_name(self):
         """
         return self._config.session_name
 
+    @property
+    def log_dir(self):
+        return self._config.log_dir
+
     @property
     def http_session(self):
         if self._http_session is None:

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

@@ -115,6 +115,32 @@ async def test_load_multiple_modules(aiohttp_client, default_module_config):
     assert await response.text() == "Hello from TestModule1"
 
 
+async def test_cached_endpoint(aiohttp_client, default_module_config):
+    """
+    Test whether the ray.dashboard.optional_utils.aiohttp_cache decorator works.
+    """
+    app = await start_http_server_app(default_module_config, [TestModule])
+    client = await aiohttp_client(app)
+
+    response = await client.get("/not_cached")
+    assert response.status == 200
+    assert await response.text() == "Hello, World from GET /not_cached, count: 1"
+
+    # Call again, count should increase.
+    response = await client.get("/not_cached")
+    assert response.status == 200
+    assert await response.text() == "Hello, World from GET /not_cached, count: 2"
+
+    response = await client.get("/cached")
+    assert response.status == 200
+    assert await response.text() == "Hello, World from GET /cached, count: 1"
+
+    # Call again, count should NOT increase.
+    response = await client.get("/cached")
+    assert response.status == 200
+    assert await response.text() == "Hello, World from GET /cached, count: 1"
+
+
 async def test_streamed_iota(aiohttp_client, default_module_config):
     # TODO(ryw): also test streams that raise exceptions.
     app = await start_http_server_app(default_module_config, [TestModule])

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

@@ -5,6 +5,7 @@
 import signal
 from typing import AsyncIterator
 
+from ray.dashboard import optional_utils
 from ray.dashboard.optional_deps import aiohttp
 
 from ray.dashboard.subprocesses.module import SubprocessModule
@@ -28,6 +29,8 @@ class TestModule(BaseTestModule):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.run_finished = False
+        self.not_cached_count = 0
+        self.cached_count = 0
 
     async def run(self):
         await super().run()
@@ -49,6 +52,21 @@ async def echo(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
         body = await req.text()
         return aiohttp.web.Response(text="Hello, World from POST /echo from " + body)
 
+    @routes.get("/not_cached")
+    async def not_cached(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
+        self.not_cached_count += 1
+        return aiohttp.web.Response(
+            text=f"Hello, World from GET /not_cached, count: {self.not_cached_count}"
+        )
+
+    @routes.get("/cached")
+    @optional_utils.aiohttp_cache
+    async def cached(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
+        self.cached_count += 1
+        return aiohttp.web.Response(
+            text=f"Hello, World from GET /cached, count: {self.cached_count}"
+        )
+
     @routes.put("/error")
     async def make_error(self, req: aiohttp.web.Request) -> aiohttp.web.Response:
         raise ValueError("This is an error")