[core][DO NOT MERGE] End-to-end testing changes for removing task events out of GCS - 2

[karticam]

test PR

contains all changes from PR 2/n to 10/n. All such PRs are prefixed by [taskEvents out of GCS].
Besides that, the other commits are:

1. 360bb91 => toggles the flag used to trigger the entire feature
   essentially these flags were configured:
   enable_ray_event: true
enable_ray_task_event_recorder: true
enable_task_events_to_dashboard_head: true
enable_core_worker_task_event_to_gcs: false
2. 899bd35 => disables some tests which are known followups for the feature.

• Unlike GcsTaskManager, we have not yet added support for metrics on task events head for events stored, events recorder etc.
• There is one edge case in test_task_failure_4.py which is yet to be covered.

[gemini-code-assist]

Code Review

This pull request implements the migration of task events out of GCS to the dashboard head. It introduces a new TaskEventsHead subprocess module, an in-memory TaskEventStorage on the dashboard head, and a background TaskEventManager to reconcile task states against GCS worker and job lifecycle events. Additionally, it integrates a new RayTaskEventRecorder in the core worker to export task events to the aggregator and updates the state API to query task events from the dashboard head. The review feedback highlights critical robustness and correctness issues, including a runtime TypeError when slice-deleting protobuf repeated fields, potential background task crashes due to unhandled gRPC and subscription exceptions, silent omission of task events lacking static metadata in query filters, and a redundant variable assignment.

[gemini-code-assist]

The current implementation of _passes_filters immediately discards any task event that does not have task_info populated. In a distributed system, lifecycle events (like RUNNING or FINISHED) can arrive out of order or definition events can be dropped due to memory limits. Discarding these events means tasks with only state updates will be silently hidden from the user entirely.

We can make this check robust by removing the strict task_info check and instead matching against the top-level job_id (which is always present on TaskEvents) and using default values for missing task_info fields.

    task_info = task_event.task_info
    if filters.exclude_driver and task_info.type == TaskType.DRIVER_TASK:
        return False
    for task_filter in filters.task_filters:
        if not _apply_predicate(
            task_filter.predicate, task_event.task_id, task_filter.task_id
        ):
            return False
    for job_filter in filters.job_filters:
        if not _apply_predicate(
            job_filter.predicate, task_event.job_id, job_filter.job_id
        ):
            return False

[gemini-code-assist]

In Python, protobuf repeated composite fields (represented as RepeatedCompositeContainer) do not support slice deletion (e.g., del container[:to_drop]). Attempting to do so will raise a TypeError at runtime, crashing the dashboard head process when a task accumulates more than MAX_NUM_PROFILE_EVENTS_PER_TASK profile events.

Instead, we should copy the kept events, clear the field, and extend it.

Suggested change

	del existing.profile_events.events[:to_drop]
	kept_events = list(existing.profile_events.events)[to_drop:]
	existing.profile_events.ClearField("events")
	existing.profile_events.events.extend(kept_events)

[gemini-code-assist]

If GCS is down or slow to start up, subscriber.subscribe() will raise an exception, causing the background subscription tasks to die permanently at startup. Additionally, if subscriber.poll() fails due to a transient network error, the loops will enter a tight busy-loop, spamming logs and consuming 100% CPU.

We should wrap the subscription step in a retry loop and add a sleep delay in the exception handlers to prevent CPU exhaustion.

    async def _subscribe_for_worker_deaths(self) -> None:
        subscriber = GcsAioWorkerDeltaSubscriber(address=self._gcs_address)
        while True:
            try:
                await subscriber.subscribe()
                break
            except Exception as e:
                logger.warning(f"Failed to subscribe to worker deaths, retrying in 1s: {e}")
                await asyncio.sleep(1)
        while True:
            try:
                for _, worker_delta in await subscriber.poll(
                    batch_size=_SUBSCRIBER_POLL_BATCH_SIZE
                ):
                    self._handle_worker_delta(worker_delta)
            except Exception:
                logger.exception("Failed handling worker-death notifications.")
                await asyncio.sleep(1)

    async def _subscribe_for_finished_jobs(self) -> None:
        subscriber = GcsAioJobSubscriber(address=self._gcs_address)
        while True:
            try:
                await subscriber.subscribe()
                break
            except Exception as e:
                logger.warning(f"Failed to subscribe to finished jobs, retrying in 1s: {e}")
                await asyncio.sleep(1)
        while True:
            try:
                for _, job_data in await subscriber.poll(
                    batch_size=_SUBSCRIBER_POLL_BATCH_SIZE
                ):
                    self._handle_job_update(job_data)
            except Exception:
                logger.exception("Failed handling job-finished notifications.")
                await asyncio.sleep(1)

[gemini-code-assist]

If GCS is temporarily unavailable or restarts, GetWorkerInfo can raise a gRPC error. This will propagate out of asyncio.gather and crash the background task. We should wrap the gRPC call in a try-except block to handle errors gracefully.

Suggested change

	async def _get_worker_info(
	self, worker_id: bytes
	) -> Optional[gcs_pb2.WorkerTableData]:
	if self._worker_info_stub is None:
	self._worker_info_stub = gcs_service_pb2_grpc.WorkerInfoGcsServiceStub(
	self._gcs_aio_channel
	)
	reply = await self._worker_info_stub.GetWorkerInfo(
	gcs_service_pb2.GetWorkerInfoRequest(worker_id=worker_id)
	)
	if not reply.HasField("worker_table_data"):
	return None
	return reply.worker_table_data
	async def _get_worker_info(
	self, worker_id: bytes
	) -> Optional[gcs_pb2.WorkerTableData]:
	if self._worker_info_stub is None:
	self._worker_info_stub = gcs_service_pb2_grpc.WorkerInfoGcsServiceStub(
	self._gcs_aio_channel
	)
	try:
	reply = await self._worker_info_stub.GetWorkerInfo(
	gcs_service_pb2.GetWorkerInfoRequest(worker_id=worker_id)
	)
	except Exception as e:
	logger.error(f"Failed to get worker info for {worker_id.hex()}: {e}")
	return None
	if not reply.HasField("worker_table_data"):
	return None
	return reply.worker_table_data

[gemini-code-assist]

This assignment is redundant because self._num_task_attempts_dropped_tracked is reassigned again on line 131 after the set is modified. We can safely remove this line.

Suggested change

	self._num_task_attempts_dropped_tracked = len(self._dropped_task_attempts)
	if num_to_evict == 0:
	return
	if num_to_evict == 0:
	return

[cursor]

Lost failures on raylet death

High Severity

With the new defaults, task events no longer go to GCS and instead flow through the node-local aggregator. When that raylet/aggregator dies, the terminal failure event can be lost because there is no GCS fallback, so tasks may never be marked failed in the dashboard-head store.

Additional Locations (1)

• src/ray/common/ray_config_def.h#L1118-L1136

 

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

[cursor]

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

There are 2 total unresolved issues (including 1 from previous review).

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

[cursor]

Stale task-events client after disconnect

Medium Severity

GlobalState.disconnect() clears _global_state_accessor but leaves _task_events_head_client cached. After reconnect, _get_task_events_head_client() keeps returning that client, which still holds the old accessor, so ray.timeline() / profile_events() can query with a stale GCS handle when the dashboard-head read path is enabled.

Additional Locations (2)

• python/ray/_private/state.py#L118-L119
• python/ray/_private/state.py#L370-L377

 

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