[core][taskEvents out of GCS][3/n] Export task events from aggregator agent to dashboard head

[karticam]

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, sinceTaskEventsMetadataBuffer.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.

[gemini-code-assist]

Code Review

This pull request replaces the GCS-based task events publisher with a new HTTP-based publisher (AsyncDashboardHeadPublisherClient) that sends events directly to the dashboard head. It also introduces a new TaskEventsHead module to receive and temporarily buffer these events in memory. Feedback on the changes suggests setting a maximum length on the in-memory event buffer (collections.deque) in TaskEventsHead to prevent potential out-of-memory (OOM) issues from unbounded growth.

[gemini-code-assist]

The in-memory buffer self._events is currently unbounded. In a long-running cluster with high task throughput, this will lead to unbounded memory growth and eventually cause the dashboard head process to be OOM-killed.

Even though this is a temporary placeholder before real storage is implemented, it is highly recommended to set a reasonable maxlen on the deque (e.g., 100000) as a defensive safeguard to prevent memory exhaustion.

Suggested change

	self._events = collections.deque()
	self._events = collections.deque(maxlen=100000)

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{Reviewed by Cursor Bugbot for commit f632c5e. Configure here.}

[cursor]

Wrong status for bad payload

Low Severity

Deserialization failures for /api/task_events return HTTPStatusCode.INTERNAL_ERROR (500). A malformed client body is a client error, so this misclassifies bad input as a server failure, unlike nearby handlers such as EventHead.report_events which treat parse failures as bad requests.

 

^{Reviewed by Cursor Bugbot for commit f632c5e. Configure here.}