[core][event] rename task_execution_event to task_lifecycle_event

Signed-off-by: Cuong Nguyen <can@anyscale.com>

Author: can-anyscale

doc/source/ray-observability/user-guides/ray-event-export.rst

@@ -45,7 +45,7 @@ For each task, Ray exports two types of events: Task Definition Event and Task E
   and actor tasks respectively.
 * Task Execution Events contain task state transition information and metadata 
   generated during task execution. 
-  See `src/ray/protobuf/public/events_task_execution_event.proto <https://github.com/ray-project/ray/blob/master/src/ray/protobuf/public/events_task_execution_event.proto>`_ for the event format.
+  See `src/ray/protobuf/public/events_task_lifecycle_event.proto <https://github.com/ray-project/ray/blob/master/src/ray/protobuf/public/events_task_lifecycle_event.proto>`_ for the event format.
 
 An example of a Task Definition Event and a Task Execution Event:
 

python/ray/dashboard/modules/aggregator/aggregator_agent.py

@@ -79,8 +79,7 @@
 # The list of all supported event types can be found in src/ray/protobuf/public/events_base_event.proto (EventType enum)
 # By default TASK_PROFILE_EVENT is not exposed to external services
 DEFAULT_EXPOSABLE_EVENT_TYPES = (
-    "TASK_DEFINITION_EVENT,TASK_EXECUTION_EVENT,"
-    "ACTOR_TASK_DEFINITION_EVENT,ACTOR_TASK_EXECUTION_EVENT,"
+    "TASK_DEFINITION_EVENT,TASK_LIFECYCLE_EVENT,ACTOR_TASK_DEFINITION_EVENT,"
     "DRIVER_JOB_DEFINITION_EVENT,DRIVER_JOB_EXECUTION_EVENT,"
     "ACTOR_DEFINITION_EVENT,ACTOR_LIFECYCLE_EVENT,"
     "NODE_DEFINITION_EVENT,NODE_LIFECYCLE_EVENT,"

python/ray/dashboard/modules/aggregator/tests/test_aggregator_agent.py

@@ -41,8 +41,8 @@
 from ray.core.generated.events_task_definition_event_pb2 import (
     TaskDefinitionEvent,
 )
-from ray.core.generated.events_task_execution_event_pb2 import (
-    TaskExecutionEvent,
+from ray.core.generated.events_task_lifecycle_event_pb2 import (
+    TaskLifecycleEvent,
 )
 from ray.core.generated.events_task_profile_events_pb2 import TaskProfileEvents
 from ray.core.generated.profile_events_pb2 import ProfileEventEntry, ProfileEvents
@@ -547,20 +547,23 @@ def _verify_task_definition_event_json(req_json, expected_timestamp):
     }
 
 
-def _create_task_execution_event_proto(timestamp):
+def _create_task_lifecycle_event_proto(timestamp):
     return RayEvent(
         event_id=b"1",
         source_type=RayEvent.SourceType.CORE_WORKER,
-        event_type=RayEvent.EventType.TASK_EXECUTION_EVENT,
+        event_type=RayEvent.EventType.TASK_LIFECYCLE_EVENT,
         timestamp=timestamp,
         severity=RayEvent.Severity.INFO,
         session_name="test_session",
-        task_execution_event=TaskExecutionEvent(
+        task_lifecycle_event=TaskLifecycleEvent(
             task_id=b"1",
             task_attempt=1,
-            task_state={
-                TaskStatus.RUNNING: timestamp,
-            },
+            state_transitions=[
+                TaskLifecycleEvent.StateTransition(
+                    state=TaskStatus.RUNNING,
+                    timestamp=timestamp,
+                ),
+            ],
             ray_error_info=RayErrorInfo(
                 error_type=ErrorType.TASK_EXECUTION_EXCEPTION,
             ),
@@ -571,13 +574,13 @@ def _create_task_execution_event_proto(timestamp):
     )
 
 
-def _verify_task_execution_event_json(req_json, expected_timestamp):
+def _verify_task_lifecycle_event_json(req_json, expected_timestamp):
     assert len(req_json) == 1
 
     # Verify the base event fields
     assert req_json[0]["eventId"] == base64.b64encode(b"1").decode()
     assert req_json[0]["sourceType"] == "CORE_WORKER"
-    assert req_json[0]["eventType"] == "TASK_EXECUTION_EVENT"
+    assert req_json[0]["eventType"] == "TASK_LIFECYCLE_EVENT"
     assert req_json[0]["timestamp"] == expected_timestamp
     assert req_json[0]["severity"] == "INFO"
     assert (
@@ -587,23 +590,26 @@ def _verify_task_execution_event_json(req_json, expected_timestamp):
 
     # Verify the task execution event specific fields
     assert (
-        req_json[0]["taskExecutionEvent"]["taskId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["taskId"] == base64.b64encode(b"1").decode()
     )
-    assert req_json[0]["taskExecutionEvent"]["taskAttempt"] == 1
-    assert req_json[0]["taskExecutionEvent"]["taskState"] == {
-        "8": expected_timestamp,
-    }
+    assert req_json[0]["taskLifecycleEvent"]["taskAttempt"] == 1
+    assert req_json[0]["taskLifecycleEvent"]["stateTransitions"] == [
+        {
+            "state": "RUNNING",
+            "timestamp": expected_timestamp,
+        }
+    ]
     assert (
-        req_json[0]["taskExecutionEvent"]["rayErrorInfo"]["errorType"]
+        req_json[0]["taskLifecycleEvent"]["rayErrorInfo"]["errorType"]
         == "TASK_EXECUTION_EXCEPTION"
     )
     assert (
-        req_json[0]["taskExecutionEvent"]["nodeId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["nodeId"] == base64.b64encode(b"1").decode()
     )
     assert (
-        req_json[0]["taskExecutionEvent"]["workerId"] == base64.b64encode(b"1").decode()
+        req_json[0]["taskLifecycleEvent"]["workerId"] == base64.b64encode(b"1").decode()
     )
-    assert req_json[0]["taskExecutionEvent"]["workerPid"] == 1
+    assert req_json[0]["taskLifecycleEvent"]["workerPid"] == 1
 
 
 def _create_profile_event_request(timestamp):
@@ -676,9 +682,9 @@ def _verify_profile_event_json(req_json, expected_timestamp):
         id="task_definition_event",
     ),
     pytest.param(
-        _create_task_execution_event_proto,
-        _verify_task_execution_event_json,
-        id="task_execution_event",
+        _create_task_lifecycle_event_proto,
+        _verify_task_lifecycle_event_json,
+        id="task_lifecycle_event",
     ),
     pytest.param(
         _create_profile_event_request, _verify_profile_event_json, id="profile_event"
@@ -693,7 +699,7 @@ def _verify_profile_event_json(req_json, expected_timestamp):
         {
             "env_vars": {
                 "RAY_DASHBOARD_AGGREGATOR_AGENT_EVENTS_EXPORT_ADDR": _EVENT_AGGREGATOR_AGENT_TARGET_ADDR,
-                "RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES": "TASK_DEFINITION_EVENT,TASK_EXECUTION_EVENT,ACTOR_TASK_DEFINITION_EVENT,ACTOR_TASK_EXECUTION_EVENT,TASK_PROFILE_EVENT",
+                "RAY_DASHBOARD_AGGREGATOR_AGENT_EXPOSABLE_EVENT_TYPES": "TASK_DEFINITION_EVENT,TASK_LIFECYCLE_EVENT,ACTOR_TASK_DEFINITION_EVENT,TASK_PROFILE_EVENT",
             },
         },
     ],

src/ray/core_worker/task_event_buffer.cc

@@ -216,17 +216,20 @@ void TaskStatusEvent::PopulateRpcRayTaskDefinitionEvent(T &definition_event_data
   }
 }
 
-void TaskStatusEvent::PopulateRpcRayTaskExecutionEvent(
-    rpc::events::TaskExecutionEvent &execution_event_data,
+void TaskStatusEvent::PopulateRpcRayTaskLifecycleEvent(
+    rpc::events::TaskLifecycleEvent &lifecycle_event_data,
     google::protobuf::Timestamp timestamp) {
   // Task identifier
-  execution_event_data.set_task_id(task_id_.Binary());
-  execution_event_data.set_task_attempt(attempt_number_);
+  lifecycle_event_data.set_task_id(task_id_.Binary());
+  lifecycle_event_data.set_task_attempt(attempt_number_);
 
   // Task state
-  auto &task_state = *execution_event_data.mutable_task_state();
   if (task_status_ != rpc::TaskStatus::NIL) {
-    task_state[task_status_] = timestamp;
+    rpc::events::TaskLifecycleEvent::StateTransition state_transition;
+    state_transition.set_state(task_status_);
+    state_transition.mutable_timestamp()->CopyFrom(timestamp);
+    *lifecycle_event_data.mutable_state_transitions()->Add() =
+        std::move(state_transition);
   }
 
   // Task property updates
@@ -235,30 +238,30 @@ void TaskStatusEvent::PopulateRpcRayTaskExecutionEvent(
   }
 
   if (state_update_->error_info_.has_value()) {
-    execution_event_data.mutable_ray_error_info()->CopyFrom(*state_update_->error_info_);
+    lifecycle_event_data.mutable_ray_error_info()->CopyFrom(*state_update_->error_info_);
   }
 
   if (state_update_->node_id_.has_value()) {
     RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)
             .WithField("TaskStatus", task_status_)
         << "Node ID should be included when task status changes to "
            "SUBMITTED_TO_WORKER.";
-    execution_event_data.set_node_id(state_update_->node_id_->Binary());
+    lifecycle_event_data.set_node_id(state_update_->node_id_->Binary());
   }
 
   if (state_update_->worker_id_.has_value()) {
     RAY_CHECK(task_status_ == rpc::TaskStatus::SUBMITTED_TO_WORKER)
             .WithField("TaskStatus", task_status_)
         << "Worker ID should be included when task status changes to "
            "SUBMITTED_TO_WORKER.";
-    execution_event_data.set_worker_id(state_update_->worker_id_->Binary());
+    lifecycle_event_data.set_worker_id(state_update_->worker_id_->Binary());
   }
 
   if (state_update_->pid_.has_value()) {
-    execution_event_data.set_worker_pid(state_update_->pid_.value());
+    lifecycle_event_data.set_worker_pid(state_update_->pid_.value());
   }
 
-  execution_event_data.set_job_id(job_id_.Binary());
+  lifecycle_event_data.set_job_id(job_id_.Binary());
 }
 
 void TaskStatusEvent::PopulateRpcRayEventBaseFields(
@@ -278,7 +281,7 @@ void TaskStatusEvent::PopulateRpcRayEventBaseFields(
       ray_event.set_event_type(rpc::events::RayEvent::TASK_DEFINITION_EVENT);
     }
   } else {
-    ray_event.set_event_type(rpc::events::RayEvent::TASK_EXECUTION_EVENT);
+    ray_event.set_event_type(rpc::events::RayEvent::TASK_LIFECYCLE_EVENT);
   }
 }
 
@@ -301,14 +304,14 @@ void TaskStatusEvent::ToRpcRayEvents(RayEventsTuple &ray_events_tuple) {
   }
 
   // Populate the task execution event
-  PopulateRpcRayEventBaseFields(ray_events_tuple.task_execution_event.has_value()
-                                    ? ray_events_tuple.task_execution_event.value()
-                                    : ray_events_tuple.task_execution_event.emplace(),
+  PopulateRpcRayEventBaseFields(ray_events_tuple.task_lifecycle_event.has_value()
+                                    ? ray_events_tuple.task_lifecycle_event.value()
+                                    : ray_events_tuple.task_lifecycle_event.emplace(),
                                 false,
                                 timestamp);
-  auto task_execution_event =
-      ray_events_tuple.task_execution_event.value().mutable_task_execution_event();
-  PopulateRpcRayTaskExecutionEvent(*task_execution_event, timestamp);
+  auto task_lifecycle_event =
+      ray_events_tuple.task_lifecycle_event.value().mutable_task_lifecycle_event();
+  PopulateRpcRayTaskLifecycleEvent(*task_lifecycle_event, timestamp);
 }
 
 void TaskProfileEvent::ToRpcTaskEvents(rpc::TaskEvents *rpc_task_events) {
@@ -629,9 +632,9 @@ TaskEventBufferImpl::CreateRayEventsDataToSend(
       auto events = data->add_events();
       *events = std::move(ray_events_tuple.task_definition_event.value());
     }
-    if (ray_events_tuple.task_execution_event) {
+    if (ray_events_tuple.task_lifecycle_event) {
       auto events = data->add_events();
-      *events = std::move(ray_events_tuple.task_execution_event.value());
+      *events = std::move(ray_events_tuple.task_lifecycle_event.value());
     }
     if (ray_events_tuple.task_profile_event) {
       auto events = data->add_events();
@@ -656,7 +659,7 @@ TaskEventBuffer::TaskEventDataToSend TaskEventBufferImpl::CreateDataToSend(
     const absl::flat_hash_set<TaskAttempt> &dropped_task_attempts_to_send) {
   // Aggregate the task events by TaskAttempt.
   absl::flat_hash_map<TaskAttempt, rpc::TaskEvents> agg_task_events;
-  // (task_attempt, (task_definition_event, task_execution_event, task_profile_event))
+  // (task_attempt, (task_definition_event, task_lifecycle_event, task_profile_event))
   absl::flat_hash_map<TaskAttempt, RayEventsTuple> agg_ray_events;
 
   auto to_rpc_event_fn =

src/ray/core_worker/task_event_buffer.h

@@ -44,15 +44,15 @@ namespace worker {
 using TaskAttempt = std::pair<TaskID, int32_t>;
 
 /// A struct containing a tuple of rpc::events::RayEvent.
-/// When converting the TaskStatusEvent, task_definition_event and task_execution_event
+/// When converting the TaskStatusEvent, task_definition_event and task_lifecycle_event
 /// will be populated with rpc::events::TaskDefinitionEvent and
-/// rpc::events::TaskExecutionEvent respectively. When converting the TaskProfileEvent,
+/// rpc::events::TaskLifecycleEvent respectively. When converting the TaskProfileEvent,
 /// task_profile_event will be populated with rpc::events::TaskProfileEvent. A struct is
-/// needed because the TaskProfileEvent, TaskDefinitionEvent and TaskExecutionEvent all
+/// needed because the TaskProfileEvent, TaskDefinitionEvent and TaskLifecycleEvent all
 /// can share the same task_id and attempt_number.
 struct RayEventsTuple {
   std::optional<rpc::events::RayEvent> task_definition_event;
-  std::optional<rpc::events::RayEvent> task_execution_event;
+  std::optional<rpc::events::RayEvent> task_lifecycle_event;
   std::optional<rpc::events::RayEvent> task_profile_event;
 };
 
@@ -171,7 +171,7 @@ class TaskStatusEvent : public TaskEvent {
 
   /// The function to convert the TaskStatusEvent class to a pair of
   /// rpc::events::RayEvent with rpc::events::TaskDefinitionEvent and
-  /// rpc::events::TaskExecutionEvent respectively. The TaskExecutionEvent will always
+  /// rpc::events::TaskLifecycleEvent respectively. The TaskLifecycleEvent will always
   /// be populated. The TaskDefinitionEvent will be populated only when the task_spec_
   /// is not null.
   /// NOTE: this method will modify internal states by moving fields of task_spec_ to
@@ -189,9 +189,9 @@ class TaskStatusEvent : public TaskEvent {
   template <typename T>
   void PopulateRpcRayTaskDefinitionEvent(T &definition_event_data);
 
-  // Helper functions to populate the task execution event of rpc::events::RayEvent
-  void PopulateRpcRayTaskExecutionEvent(
-      rpc::events::TaskExecutionEvent &execution_event_data,
+  // Helper functions to populate the task lifecycle event of rpc::events::RayEvent
+  void PopulateRpcRayTaskLifecycleEvent(
+      rpc::events::TaskLifecycleEvent &lifecycle_event_data,
       google::protobuf::Timestamp timestamp);
 
   // Helper functions to populate the base fields of rpc::events::RayEvent

src/ray/core_worker/tests/task_event_buffer_test.cc

@@ -439,9 +439,9 @@ TEST_P(TaskEventBufferTestDifferentDestination, TestFlushEvents) {
       auto new_event = expected_ray_events_data.add_events();
       *new_event = std::move(ray_events_tuple.task_definition_event.value());
     }
-    if (ray_events_tuple.task_execution_event) {
+    if (ray_events_tuple.task_lifecycle_event) {
       auto new_event = expected_ray_events_data.add_events();
-      *new_event = std::move(ray_events_tuple.task_execution_event.value());
+      *new_event = std::move(ray_events_tuple.task_lifecycle_event.value());
     }
     if (ray_events_tuple.task_profile_event) {
       auto new_event = expected_ray_events_data.add_events();
@@ -757,9 +757,9 @@ TEST_P(TaskEventBufferTestLimitBufferDifferentDestination,
       auto new_event = expected_ray_events_data.add_events();
       *new_event = std::move(ray_events_tuple.task_definition_event.value());
     }
-    if (ray_events_tuple.task_execution_event) {
+    if (ray_events_tuple.task_lifecycle_event) {
       auto new_event = expected_ray_events_data.add_events();
-      *new_event = std::move(ray_events_tuple.task_execution_event.value());
+      *new_event = std::move(ray_events_tuple.task_lifecycle_event.value());
     }
     if (ray_events_tuple.task_profile_event) {
       auto new_event = expected_ray_events_data.add_events();
@@ -960,7 +960,7 @@ TEST_F(TaskEventBufferTest, TestTaskProfileEventToRpcRayEvents) {
   // Verify that the second event is nullopt (empty)
   EXPECT_FALSE(ray_events_tuple.task_definition_event.has_value())
       << "TaskProfileEvent should be populated in RayEventsTuple";
-  EXPECT_FALSE(ray_events_tuple.task_execution_event.has_value())
+  EXPECT_FALSE(ray_events_tuple.task_lifecycle_event.has_value())
       << "TaskProfileEvent should be populated in RayEventsTuple";
 
   // Verify that the first event contains the profile event
@@ -1086,9 +1086,9 @@ TEST_P(TaskEventBufferTestDifferentDestination,
     auto new_event = expected_ray_events_data.add_events();
     *new_event = std::move(ray_events_tuple.task_definition_event.value());
   }
-  if (ray_events_tuple.task_execution_event) {
+  if (ray_events_tuple.task_lifecycle_event) {
     auto new_event = expected_ray_events_data.add_events();
-    *new_event = std::move(ray_events_tuple.task_execution_event.value());
+    *new_event = std::move(ray_events_tuple.task_lifecycle_event.value());
   }
   if (ray_events_tuple.task_profile_event) {
     auto new_event = expected_ray_events_data.add_events();

src/ray/gcs/gcs_ray_event_converter.cc

@@ -37,8 +37,8 @@ GcsRayEventConverter::ConvertToTaskEventDataRequests(
       task_event = ConvertToTaskEvents(std::move(*event.mutable_task_definition_event()));
       break;
     }
-    case rpc::events::RayEvent::TASK_EXECUTION_EVENT: {
-      task_event = ConvertToTaskEvents(std::move(*event.mutable_task_execution_event()));
+    case rpc::events::RayEvent::TASK_LIFECYCLE_EVENT: {
+      task_event = ConvertToTaskEvents(std::move(*event.mutable_task_lifecycle_event()));
       break;
     }
     case rpc::events::RayEvent::TASK_PROFILE_EVENT: {
@@ -144,7 +144,7 @@ rpc::TaskEvents GcsRayEventConverter::ConvertToTaskEvents(
 }
 
 rpc::TaskEvents GcsRayEventConverter::ConvertToTaskEvents(
-    rpc::events::TaskExecutionEvent &&event) {
+    rpc::events::TaskLifecycleEvent &&event) {
   rpc::TaskEvents task_event;
   task_event.set_task_id(event.task_id());
   task_event.set_attempt_number(event.task_attempt());
@@ -156,9 +156,9 @@ rpc::TaskEvents GcsRayEventConverter::ConvertToTaskEvents(
   task_state_update->set_worker_pid(event.worker_pid());
   task_state_update->mutable_error_info()->Swap(event.mutable_ray_error_info());
 
-  for (const auto &[state, timestamp] : event.task_state()) {
-    int64_t ns = ProtoTimestampToAbslTimeNanos(timestamp);
-    (*task_state_update->mutable_state_ts_ns())[state] = ns;
+  for (const auto &state_transition : event.state_transitions()) {
+    int64_t ns = ProtoTimestampToAbslTimeNanos(state_transition.timestamp());
+    (*task_state_update->mutable_state_ts_ns())[state_transition.state()] = ns;
   }
   return task_event;
 }

src/ray/gcs/gcs_ray_event_converter.h

@@ -51,11 +51,11 @@ class GcsRayEventConverter {
   /// \return The output TaskEvents to populate.
   rpc::TaskEvents ConvertToTaskEvents(rpc::events::TaskProfileEvents &&event);
 
-  /// Convert a TaskExecutionEvent to a TaskEvents.
+  /// Convert a TaskLifecycleEvent to a TaskEvents.
   ///
-  /// \param event The TaskExecutionEvent to convert.
+  /// \param event The TaskLifecycleEvent to convert.
   /// \return The output TaskEvents to populate.
-  rpc::TaskEvents ConvertToTaskEvents(rpc::events::TaskExecutionEvent &&event);
+  rpc::TaskEvents ConvertToTaskEvents(rpc::events::TaskLifecycleEvent &&event);
 
   /// Convert an ActorTaskDefinitionEvent to a TaskEvents.
   ///
@@ -99,7 +99,7 @@ class GcsRayEventConverter {
       std::vector<rpc::AddTaskEventDataRequest> &requests_per_job_id,
       absl::flat_hash_map<std::string, size_t> &job_id_to_index);
 
-  FRIEND_TEST(GcsRayEventConverterTest, TestConvertTaskExecutionEvent);
+  FRIEND_TEST(GcsRayEventConverterTest, TestConvertTaskLifecycleEvent);
   FRIEND_TEST(GcsRayEventConverterTest, TestConvertActorTaskDefinitionEvent);
 };