[core] fix ray_event_recorder initialization

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

Author: can-anyscale

src/ray/gcs/gcs_server.cc

@@ -296,7 +296,7 @@ void GcsServer::DoStart(const GcsInitData &gcs_init_data) {
   // Init metrics and event exporter.
   metrics_agent_client_->WaitForServerReady([this](const Status &server_status) {
     stats::InitOpenTelemetryExporter(config_.metrics_agent_port, server_status);
-    ray_event_recorder_->StartExportingEvents();
+    ray_event_recorder_->StartExportingEvents(server_status);
   });
 
   // Start RPC server when all tables have finished loading initial

src/ray/observability/fake_ray_event_recorder.h

@@ -24,7 +24,7 @@ namespace observability {
 
 class FakeRayEventRecorder : public RayEventRecorderInterface {
  public:
-  void StartExportingEvents() override {}
+  void StartExportingEvents(const Status &aggregator_agent_status) override {}
   void AddEvents(std::vector<std::unique_ptr<RayEventInterface>> &&data_list) override {
     absl::MutexLock lock(&mutex_);
     buffer_.insert(buffer_.end(),

src/ray/observability/ray_event_recorder.cc

@@ -34,15 +34,22 @@ RayEventRecorder::RayEventRecorder(
       buffer_(max_buffer_size),
       dropped_events_counter_(dropped_events_counter) {}
 
-void RayEventRecorder::StartExportingEvents() {
+void RayEventRecorder::StartExportingEvents(const Status &aggregator_agent_status) {
   absl::MutexLock lock(&mutex_);
+  if (!aggregator_agent_status.ok()) {
+    RAY_LOG(ERROR) << "Failed to establish connection to the event aggregator agent. "
+                   << "Error: " << aggregator_agent_status.ToString();
+    return;
+  }
   RAY_CHECK(!exporting_started_)
       << "RayEventRecorder::StartExportingEvents() should be called only once.";
   exporting_started_ = true;
   periodical_runner_->RunFnPeriodically(
       [this]() { ExportEvents(); },
       RayConfig::instance().ray_events_report_interval_ms(),
       "RayEventRecorder.ExportEvents");
+  RAY_LOG(INFO) << "Successfully established connection to the event aggregator agent. "
+                   "Events will start being exported.";
 }
 
 void RayEventRecorder::ExportEvents() {

src/ray/observability/ray_event_recorder_interface.h

@@ -17,6 +17,7 @@
 #include <memory>
 #include <vector>
 
+#include "ray/common/status.h"
 #include "ray/observability/ray_event_interface.h"
 
 namespace ray {
@@ -29,7 +30,7 @@ class RayEventRecorderInterface {
   // Start exporting events to the event aggregator by periodically sending events to
   // the event aggregator. This should be called only once. Subsequent calls will be
   // ignored.
-  virtual void StartExportingEvents() = 0;
+  virtual void StartExportingEvents(const Status &aggregator_agent_status) = 0;
 
   // Add a vector of data to the internal buffer. Data in the buffer will be sent to
   // the event aggregator periodically.

src/ray/rpc/metrics_agent_client.cc

@@ -50,7 +50,8 @@ void MetricsAgentClientImpl::WaitForServerReadyWithRetry(
                 }
               });
   if (retry_count >= max_retry) {
-    init_exporter_fn(Status::RpcError("The metrics agent server is not ready.", 14));
+    init_exporter_fn(
+        Status::RpcError("Running out of retries to initialize the metrics agent.", 14));
     return;
   }
   retry_count++;