[core][onevent] implement even merge logic at export time (#58070)

RayEvent provides a special API, merge, which allows multiple events to
be combined into a single event. This reduces gRPC message size, network
bandwidth usage, and is essential for scaling task event exports. This
PR leverages that feature.

Specifically, it clusters events into groups based on (i) entity ID and
(ii) event type. Each group is merged into a single event, which is then
added to the gRPC message body. The EntityId is a user-defined function,
implemented by the event class creator, that determines which events can
be safely merged.

```
Note: this is a redo of #56558 which gets converted because it randomize the order the events that get exported, lead to flaky tests etc. This attempt maintain the order even after merging.
```

Test:
- CI

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

Author: can-anyscale

src/ray/observability/ray_event_recorder.cc

@@ -56,9 +56,23 @@ void RayEventRecorder::ExportEvents() {
   }
   rpc::events::AddEventsRequest request;
   rpc::events::RayEventsData ray_event_data;
-  // TODO(#56391): To further optimize the performance, we can merge multiple
-  // events with the same resource ID into a single event.
+  // group the event in the buffer_ by their entity id and type; then for each group,
+  // merge the events into a single event. maintain the order of the events in the buffer.
+  std::list<std::unique_ptr<RayEventInterface>> grouped_events;
+  absl::flat_hash_map<RayEventKey,
+                      std::list<std::unique_ptr<RayEventInterface>>::iterator>
+      event_key_to_iterator;
   for (auto &event : buffer_) {
+    auto key = std::make_pair(event->GetEntityId(), event->GetEventType());
+    auto [it, inserted] = event_key_to_iterator.try_emplace(key);
+    if (inserted) {
+      grouped_events.push_back(std::move(event));
+      event_key_to_iterator[key] = std::prev(grouped_events.end());
+    } else {
+      (*it->second)->Merge(std::move(*event));
+    }
+  }
+  for (auto &event : grouped_events) {
     rpc::events::RayEvent ray_event = std::move(*event).Serialize();
     *ray_event_data.mutable_events()->Add() = std::move(ray_event);
   }

src/ray/observability/ray_event_recorder.h

@@ -53,6 +53,8 @@ class RayEventRecorder : public RayEventRecorderInterface {
   void AddEvents(std::vector<std::unique_ptr<RayEventInterface>> &&data_list);
 
  private:
+  using RayEventKey = std::pair<std::string, rpc::events::RayEvent::EventType>;
+
   rpc::EventAggregatorClient &event_aggregator_client_;
   std::shared_ptr<PeriodicalRunner> periodical_runner_;
   // Lock for thread safety when modifying the buffer.

src/ray/observability/tests/ray_event_recorder_test.cc

@@ -78,6 +78,37 @@ class RayEventRecorderTest : public ::testing::Test {
   size_t max_buffer_size_ = 5;
 };
 
+TEST_F(RayEventRecorderTest, TestMergeEvents) {
+  RayConfig::instance().initialize(
+      R"(
+{
+"enable_ray_event": true
+}
+)");
+  recorder_->StartExportingEvents();
+  rpc::JobTableData data;
+  data.set_job_id("test_job_id");
+
+  std::vector<std::unique_ptr<RayEventInterface>> events;
+  events.push_back(std::make_unique<RayDriverJobLifecycleEvent>(
+      data, rpc::events::DriverJobLifecycleEvent::CREATED, "test_session_name"));
+  events.push_back(std::make_unique<RayDriverJobLifecycleEvent>(
+      data, rpc::events::DriverJobLifecycleEvent::FINISHED, "test_session_name"));
+  recorder_->AddEvents(std::move(events));
+  io_service_.run_one();
+
+  std::vector<rpc::events::RayEvent> recorded_events = fake_client_->GetRecordedEvents();
+  // Only one event should be recorded because the two events are merged into one.
+  ASSERT_EQ(recorded_events.size(), 1);
+  ASSERT_EQ(recorded_events[0].source_type(), rpc::events::RayEvent::GCS);
+  ASSERT_EQ(recorded_events[0].session_name(), "test_session_name");
+  auto state_transitions =
+      recorded_events[0].driver_job_lifecycle_event().state_transitions();
+  ASSERT_EQ(state_transitions.size(), 2);
+  ASSERT_EQ(state_transitions[0].state(), rpc::events::DriverJobLifecycleEvent::CREATED);
+  ASSERT_EQ(state_transitions[1].state(), rpc::events::DriverJobLifecycleEvent::FINISHED);
+}
+
 TEST_F(RayEventRecorderTest, TestRecordEvents) {
   RayConfig::instance().initialize(
       R"(