[cherry-pick][2.58.0][core][taskEvents out of GCS][2/n] Replace task event buffer with ray event recorder (#64835) by elliot-barn · Pull Request #65274 · ray-project/ray

elliot-barn · 2026-08-07T01:51:33Z

Cherry-pick of #64835 (merge commit d126844) into releases/2.58.0.

Applied cleanly with no conflicts.

Not a duplicate: no existing open PR against releases/2.58.0 contains this change. Cherry-picked with AI assistance (Claude Code); pre-commit hooks ran and passed on the commit.

🤖 Generated with Claude Code

… event recorder (#64835) This PR builds on top of #64168. It removes `task_event_buffer` and replaces it with `ray_task_event_recorder`. **CHANGES**: 1. Added `RayTaskDefinitionEvent`, `RayActorTaskDefinitionEvent`, `RayTaskLifecycleEvent`, and `RayTaskProfileEvent` classes which extend RayEventInterface to be recorded by `RayTaskEventRecorder`. 2. Constructs `RayTaskEventRecorder` in `CoreWorkerProcessImpl::CreateCoreWorker`, and uses it in core worker constructor. Wires the recorder to other classes as required (`TaskManager`, `TaskReceiver`, ActorExecutionQueues etc.) 3. Uses its `AddEvents` function in parallel to where `task_event_buffer_->AddTaskEvent` is done. 4. Adds a function `worker::RecordTaskStatusEventToRecorderIfNeeded` in parallel to `task_event_buffer_.RecordTaskStatusEventIfNeeded` with similar gates. 5. Adds a flag `enable_ray_task_event_recorder` as a config variable to decide whether to use `RayTaskEventRecorder` for sending task events to aggregator agent as follows: - The event framework is used only when `enable_ray_task_event_recorder` and `enable_ray_event` are both set. - If both are set, `task_event_buffer` -> aggregator agent path is disabled. - If either is not set, `task_event_buffer` -> aggregator agent path is enabled/disabled based on the existing flag. **TESTING**: To test correct working of ray task event recorder, I turned changed the apt flags and triggered the CI suite in this PR: #65016 Commit where I change flags: ba61a94 --------- Signed-off-by: Kartica Modi <karticamodi@gmail.com> (cherry picked from commit d126844) Signed-off-by: elliot-barn <elliot.barnwell@anyscale.com>

gemini-code-assist

Code Review

This pull request integrates the RayTaskEventRecorder path to record and export task definition, lifecycle, and profile events to the event aggregator, while disabling the legacy TaskEventBuffer aggregator path when active to prevent double reporting. The review feedback highlights several critical improvements: a potential null pointer dereference in CoreWorkerProcessImpl if the recorder is dynamically enabled, the risk of crashing the process via RAY_CHECK(false) on duplicate observability events, and an anti-pattern of using std::optional which prevents moving the TaskStateUpdate parameter and causes unnecessary copies.

gemini-code-assist · 2026-08-07T01:53:33Z

+      PeriodicalRunner::Create(observability::RayTaskEventRecorder::Enabled()
+                                   ? ray_task_event_recorder_io_context_->GetIoService()
+                                   : io_service_),


Potential null pointer dereference. If observability::RayTaskEventRecorder::Enabled() is dynamically enabled in tests after CoreWorkerProcessImpl is constructed, ray_task_event_recorder_io_context_ will be nullptr, causing a crash when calling GetIoService(). We should check that ray_task_event_recorder_io_context_ is not null before accessing it.

Suggested change

PeriodicalRunner::Create(observability::RayTaskEventRecorder::Enabled()

? ray_task_event_recorder_io_context_->GetIoService()

: io_service_),

PeriodicalRunner::Create((observability::RayTaskEventRecorder::Enabled() && ray_task_event_recorder_io_context_ != nullptr)

? ray_task_event_recorder_io_context_->GetIoService()

: io_service_),

gemini-code-assist · 2026-08-07T01:53:33Z

+void RayActorTaskDefinitionEvent::MergeData(
+    RayEvent<rpc::events::ActorTaskDefinitionEvent> &&other) {
+  RAY_CHECK(false) << absl::StrFormat(
+      "MergeData called on actor task definition event for task %s attempt %d; only "
+      "one definition event is expected per task attempt.",
+      absl::BytesToHexString(data_.task_id()),
+      data_.task_attempt());
+}


Crashing the entire process with RAY_CHECK(false) because of a duplicate observability event is extremely risky in production. Observability and metrics code should be fail-safe and never crash the main application. If a duplicate event is received, we should log a warning or make it a no-op instead of crashing.

Suggested change

void RayActorTaskDefinitionEvent::MergeData(

RayEvent<rpc::events::ActorTaskDefinitionEvent> &&other) {

RAY_CHECK(false) << absl::StrFormat(

"MergeData called on actor task definition event for task %s attempt %d; only "

"one definition event is expected per task attempt.",

absl::BytesToHexString(data_.task_id()),

data_.task_attempt());

}

void RayActorTaskDefinitionEvent::MergeData(

RayEvent<rpc::events::ActorTaskDefinitionEvent> &&other) {

RAY_LOG_EVERY_N(WARNING, 100) << absl::StrFormat(

"MergeData called on actor task definition event for task %s attempt %d; only "

"one definition event is expected per task attempt.",

absl::BytesToHexString(data_.task_id()),

data_.task_attempt());

}

gemini-code-assist · 2026-08-07T01:53:33Z

+void RayTaskDefinitionEvent::MergeData(
+    RayEvent<rpc::events::TaskDefinitionEvent> &&other) {
+  RAY_CHECK(false) << absl::StrFormat(
+      "MergeData called on task definition event for task %s attempt %d; only one "
+      "definition event is expected per task attempt.",
+      absl::BytesToHexString(data_.task_id()),
+      data_.task_attempt());
+}


Crashing the entire process with RAY_CHECK(false) because of a duplicate observability event is extremely risky in production. Observability and metrics code should be fail-safe and never crash the main application. If a duplicate event is received, we should log a warning or make it a no-op instead of crashing.

Suggested change

void RayTaskDefinitionEvent::MergeData(

RayEvent<rpc::events::TaskDefinitionEvent> &&other) {

RAY_CHECK(false) << absl::StrFormat(

"MergeData called on task definition event for task %s attempt %d; only one "

"definition event is expected per task attempt.",

absl::BytesToHexString(data_.task_id()),

data_.task_attempt());

}

void RayTaskDefinitionEvent::MergeData(

RayEvent<rpc::events::TaskDefinitionEvent> &&other) {

RAY_LOG_EVERY_N(WARNING, 100) << absl::StrFormat(

"MergeData called on task definition event for task %s attempt %d; only one "

"definition event is expected per task attempt.",

absl::BytesToHexString(data_.task_id()),

data_.task_attempt());

}

gemini-code-assist · 2026-08-07T01:53:33Z

+    const std::string &session_name,
+    const NodeID &node_id,
+    bool include_task_info = false,
+    std::optional<const TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);


Using const inside std::optional (i.e., std::optional<const T>) is an anti-pattern because it prevents moving the contained value. This causes an unnecessary copy of TaskStateUpdate when passing it to the TaskStatusEvent constructor. Removing const allows the value to be moved efficiently.

Suggested change

std::optional<const TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);

std::optional<TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);

gemini-code-assist · 2026-08-07T01:53:33Z

+void RecordTaskStatusEventToRecorderIfNeeded(
+    ray::observability::RayEventRecorderInterface &ray_task_event_recorder,
+    const TaskID &task_id,
+    const JobID &job_id,
+    int32_t attempt_number,
+    const TaskSpecification &spec,
+    rpc::TaskStatus status,
+    int64_t timestamp,
+    const std::string &session_name,
+    const NodeID &node_id,
+    bool include_task_info,
+    std::optional<const TaskStatusEvent::TaskStateUpdate> state_update) {


Remove const from std::optional to allow moving the TaskStateUpdate parameter instead of copying it.

Suggested change

void RecordTaskStatusEventToRecorderIfNeeded(

ray::observability::RayEventRecorderInterface &ray_task_event_recorder,

const TaskID &task_id,

const JobID &job_id,

int32_t attempt_number,

const TaskSpecification &spec,

rpc::TaskStatus status,

int64_t timestamp,

const std::string &session_name,

const NodeID &node_id,

bool include_task_info,

std::optional<const TaskStatusEvent::TaskStateUpdate> state_update) {

void RecordTaskStatusEventToRecorderIfNeeded(

ray::observability::RayEventRecorderInterface &ray_task_event_recorder,

const TaskID &task_id,

const JobID &job_id,

int32_t attempt_number,

const TaskSpecification &spec,

rpc::TaskStatus status,

int64_t timestamp,

const std::string &session_name,

const NodeID &node_id,

bool include_task_info,

std::optional<TaskStatusEvent::TaskStateUpdate> state_update) {

cursor

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

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

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

cursor · 2026-08-07T01:56:05Z

+    const std::string &session_name,
+    const NodeID &node_id,
+    bool include_task_info = false,
+    std::optional<const TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);


Missing Doxygen on C++ APIs

Low Severity

⚠️ Document functions and classes with Doxygen /** ... */ block comments using @ tags (@brief, and @param/@return where applicable).

Additional Locations (2)

src/ray/core_worker/profile_event.h#L34-L40

src/ray/core_worker/task_event_buffer.h#L433-L435

^{Triggered by project rule: Bugbot Rules}

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

elliot-barn requested a review from a team as a code owner August 7, 2026 01:51

elliot-barn mentioned this pull request Aug 7, 2026

[cherry-pick][2.58.0][core][taskEvents out of GCS][3/n] Export task events from aggregator agent to dashboard head (#65028) #65275

Open

gemini-code-assist Bot reviewed Aug 7, 2026

View reviewed changes

cursor Bot reviewed Aug 7, 2026

View reviewed changes

ray-gardener Bot added core Issues that should be addressed in Ray Core release-test release test labels Aug 7, 2026

Navigation Menu

Uh oh!

[cherry-pick][2.58.0][core][taskEvents out of GCS][2/n] Replace task event buffer with ray event recorder (#64835) - #65274

[cherry-pick][2.58.0][core][taskEvents out of GCS][2/n] Replace task event buffer with ray event recorder (#64835)#65274
elliot-barn wants to merge 1 commit into
releases/2.58.0from
elliot-barn/cherry-pick-2.58.0-task-event-recorder-64835

elliot-barn commented Aug 7, 2026

Uh oh!

gemini-code-assist Bot left a comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Uh oh!

cursor Bot left a comment

Uh oh!

cursor Bot Aug 7, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

	std::optional<const TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);
	std::optional<TaskStatusEvent::TaskStateUpdate> state_update = std::nullopt);

Uh oh!

Conversation

elliot-barn commented Aug 7, 2026

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

cursor Bot left a comment

Choose a reason for hiding this comment

Uh oh!

cursor Bot Aug 7, 2026

Choose a reason for hiding this comment

Missing Doxygen on C++ APIs

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants