[core][event] feature flag for enabling ray export event (#57999)

We have a feature flag to control the rolling out of ray export event,
but the feature flag is missing the controlling of
`StartExportingEvents`. This PR fixes the issue.

Test:
- CI

Signed-off-by: Cuong Nguyen <[email protected]>

Author: can-anyscale

src/ray/observability/BUILD.bazel

@@ -166,6 +166,7 @@ ray_cc_library(
         ":ray_event",
         ":ray_event_recorder_interface",
         "//src/ray/common:asio",
+        "//src/ray/common:ray_config",
         "//src/ray/protobuf:events_event_aggregator_service_cc_proto",
         "//src/ray/rpc:event_aggregator_client",
         "//src/ray/util:logging",

src/ray/observability/ray_event_recorder.cc

@@ -36,6 +36,10 @@ RayEventRecorder::RayEventRecorder(
 
 void RayEventRecorder::StartExportingEvents() {
   absl::MutexLock lock(&mutex_);
+  if (!RayConfig::instance().enable_ray_event()) {
+    RAY_LOG(INFO) << "Ray event recording is disabled. Skipping start exporting events.";
+    return;
+  }
   RAY_CHECK(!exporting_started_)
       << "RayEventRecorder::StartExportingEvents() should be called only once.";
   exporting_started_ = true;
@@ -74,6 +78,9 @@ void RayEventRecorder::ExportEvents() {
 void RayEventRecorder::AddEvents(
     std::vector<std::unique_ptr<RayEventInterface>> &&data_list) {
   absl::MutexLock lock(&mutex_);
+  if (!RayConfig::instance().enable_ray_event()) {
+    return;
+  }
   if (data_list.size() + buffer_.size() > max_buffer_size_) {
     size_t events_to_remove = data_list.size() + buffer_.size() - max_buffer_size_;
     // Record dropped events from the buffer

src/ray/observability/ray_event_recorder.h

@@ -18,6 +18,7 @@
 
 #include "absl/synchronization/mutex.h"
 #include "ray/common/asio/periodical_runner.h"
+#include "ray/common/ray_config.h"
 #include "ray/observability/metric_interface.h"
 #include "ray/observability/ray_event_interface.h"
 #include "ray/observability/ray_event_recorder_interface.h"

src/ray/observability/tests/ray_event_recorder_test.cc

@@ -69,7 +69,6 @@ class RayEventRecorderTest : public ::testing::Test {
                                                    max_buffer_size_,
                                                    "gcs",
                                                    *fake_dropped_events_counter_);
-    recorder_->StartExportingEvents();
   }
 
   instrumented_io_context io_service_;
@@ -80,6 +79,13 @@ class RayEventRecorderTest : public ::testing::Test {
 };
 
 TEST_F(RayEventRecorderTest, TestRecordEvents) {
+  RayConfig::instance().initialize(
+      R"(
+{
+"enable_ray_event": true
+}
+)");
+  recorder_->StartExportingEvents();
   rpc::JobTableData data1;
   data1.set_job_id("test_job_id_1");
   data1.set_is_dead(false);
@@ -173,6 +179,13 @@ TEST_F(RayEventRecorderTest, TestRecordEvents) {
 }
 
 TEST_F(RayEventRecorderTest, TestDropEvents) {
+  RayConfig::instance().initialize(
+      R"(
+{
+"enable_ray_event": true
+}
+)");
+  recorder_->StartExportingEvents();
   size_t expected_num_dropped_events = 3;
 
   // Add more events than the buffer size
@@ -204,5 +217,31 @@ TEST_F(RayEventRecorderTest, TestDropEvents) {
   ASSERT_EQ(num_dropped_events, expected_num_dropped_events);
 }
 
+TEST_F(RayEventRecorderTest, TestDisabled) {
+  RayConfig::instance().initialize(
+      R"(
+{
+  "enable_ray_event": false
+}
+  )");
+  recorder_->StartExportingEvents();
+  rpc::JobTableData data;
+  data.set_job_id("test_job_id_1");
+  data.set_is_dead(false);
+  data.set_driver_pid(12345);
+  data.set_start_time(absl::ToUnixSeconds(absl::Now()));
+  data.set_end_time(0);
+  data.set_entrypoint("python test_script.py");
+  data.mutable_driver_address()->set_ip_address("127.0.0.1");
+
+  std::vector<std::unique_ptr<RayEventInterface>> events;
+  events.push_back(
+      std::make_unique<RayDriverJobDefinitionEvent>(data, "test_session_name"));
+  recorder_->AddEvents(std::move(events));
+  io_service_.run_one();
+  std::vector<rpc::events::RayEvent> recorded_events = fake_client_->GetRecordedEvents();
+  ASSERT_EQ(recorded_events.size(), 0);
+}
+
 }  // namespace observability
 }  // namespace ray