[cherry-pick][2.58.0][core][taskEvents out of GCS][3/n] Export task events from aggregator agent to dashboard head (#65028) - #65275
Conversation
… agent to dashboard head (#65028) Part of the effort to move task-event observability data out of GCS and onto the dashboard head. This PR adds support for moving task events from aggregator agent to dashboard head. Changes are as follows: 1. Adds a new module `task_events_head` that hosts a POST `/api/task_events` endpoint to receive task events from the aggregator agent. For now it just buffers received events in memory. More stuff to be done in upcoming PRs. 2. Adds a new publisher client `AsyncDashboardHeadPublisherClient` and a corresponding flag to `PUBLISH_EVENTS_TO_DASHBOARD_HEAD`. 3. The existing publisher `AsyncGCSTaskEventsPublisherClient` is still kept since some tests depend on it. This will be removed later after the entire migration completes. 4. The new publisher client is similar to `AsyncGCSTaskEventsPublisherClient` in that it has the same event selection to export and has the same proto building, but sends the request over HTTP POST instead of a gRPC call. 5. Each of the two publishers gets its own dropped-task-attempts metadata buffer, since`TaskEventsMetadataBuffer.get()` is destructive and a shared buffer would split the metadata across publishers. 6. The publisher resolves the dashboard head's address lazily from InternalKV (DASHBOARD_ADDRESS) via `gcs_client`, then makes the POST request. If the address isn't registered yet the publish is marked unsuccessful and the publisher retries on its next cycle; once resolved, the endpoint is cached. --------- Signed-off-by: Kartica Modi <karticamodi@gmail.com> (cherry picked from commit 3ca1915) Signed-off-by: elliot-barn <elliot.barnwell@anyscale.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the capability to migrate task events from GCS to the dashboard head. It adds a new configuration flag enable_task_events_to_dashboard_head, implements AsyncDashboardHeadPublisherClient to publish task events over HTTP, and introduces TaskEventsHead to receive and buffer these events. Feedback on the changes highlights two key improvements: setting a maxlen on the collections.deque in TaskEventsHead to prevent unbounded memory growth and potential OOM crashes, and catching the broader aiohttp.ClientError exception to ensure the cached endpoint is cleared on HTTP response errors.
| # TODO(karticam): Replace this with an in-memory store of task events. | ||
| # This will mimic current GcsTaskManager and will power state API. | ||
| # Will be done in future PRs. | ||
| self._events = collections.deque() |
There was a problem hiding this comment.
The collections.deque is initialized without a maxlen parameter. Since TaskEventsHead receives task events continuously from all nodes in the cluster and appends them to self._events indefinitely, this will lead to unbounded memory growth and eventual Out-Of-Memory (OOM) crashes of the dashboard head process.
Consider setting a sensible maxlen (e.g., 100000 to match the default GCS task events limit) to ensure memory usage remains bounded.
| self._events = collections.deque() | |
| self._events = collections.deque(maxlen=100000) |
| except (aiohttp.ClientConnectionError, asyncio.TimeoutError): | ||
| # Couldn't reach the endpoint; the dashboard head may have restarted at a | ||
| # new address. Drop the cached endpoint so the next publish re-resolves it | ||
| # from InternalKV. | ||
| self._endpoint = None | ||
| raise |
There was a problem hiding this comment.
Currently, the cached self._endpoint is only cleared when aiohttp.ClientConnectionError or asyncio.TimeoutError is raised. However, if the dashboard head restarts or is redeployed, the old endpoint might return HTTP errors (e.g., 404 Not Found or 503 Service Unavailable) via resp.raise_for_status(), which raises aiohttp.ClientResponseError.
Catching the broader aiohttp.ClientError (which is the base class for both connection and response errors) ensures that the cached endpoint is also cleared on HTTP response errors, allowing the client to re-resolve the correct endpoint from GCS on the next publish attempt.
| except (aiohttp.ClientConnectionError, asyncio.TimeoutError): | |
| # Couldn't reach the endpoint; the dashboard head may have restarted at a | |
| # new address. Drop the cached endpoint so the next publish re-resolves it | |
| # from InternalKV. | |
| self._endpoint = None | |
| raise | |
| except (aiohttp.ClientError, asyncio.TimeoutError): | |
| # Couldn't reach the endpoint or received an HTTP error; the dashboard head may | |
| # have restarted at a new address. Drop the cached endpoint so the next publish | |
| # re-resolves it from InternalKV. | |
| self._endpoint = None | |
| raise |
2 participants
Cherry-pick of #65028 (merge commit 3ca1915) into
releases/2.58.0.One contextual conflict in
src/ray/common/ray_config_def.h: on master, #64835 added itsenable_ray_task_event_recorderflag adjacent to this PR'senable_task_events_to_dashboard_headflag. Resolved by inserting only this PR's flag, so this cherry-pick is independent of #65274 (the #64835 pick) and the two can merge in either order. All other files applied cleanly.Not a duplicate: no existing open PR against
releases/2.58.0contains this change. Cherry-picked with AI assistance (Claude Code); pre-commit hooks ran and passed on the commit.🤖 Generated with Claude Code