[core][event] fix "RayEventRecorder::StartExportingEvents() should be called only once."

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

Author: can-anyscale

src/ray/gcs/gcs_server.cc

@@ -296,7 +296,13 @@ 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(server_status);
+    if (server_status.ok()) {
+      ray_event_recorder_->StartExportingEvents();
+    } else {
+      RAY_LOG(ERROR) << "Failed to establish connection to the event exporter. Events "
+                        "will not be exported. "
+                     << "Event exporter status: " << server_status.ToString();
+    }
   });
 
   // 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(const Status &aggregator_agent_status) override {}
+  void StartExportingEvents() 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,22 +34,15 @@ RayEventRecorder::RayEventRecorder(
       buffer_(max_buffer_size),
       dropped_events_counter_(dropped_events_counter) {}
 
-void RayEventRecorder::StartExportingEvents(const Status &aggregator_agent_status) {
+void RayEventRecorder::StartExportingEvents() {
   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,7 +17,6 @@
 #include <memory>
 #include <vector>
 
-#include "ray/common/status.h"
 #include "ray/observability/ray_event_interface.h"
 
 namespace ray {
@@ -30,7 +29,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(const Status &aggregator_agent_status) = 0;
+  virtual void StartExportingEvents() = 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

@@ -41,31 +41,31 @@ void MetricsAgentClientImpl::WaitForServerReadyWithRetry(
     // Only log the first time we start the retry loop.
     RAY_LOG(INFO) << "Initializing exporter ...";
   }
-  HealthCheck(rpc::HealthCheckRequest(),
-              [this, init_exporter_fn](auto &status, auto &&reply) {
-                if (status.ok() && !exporter_initialized_) {
-                  init_exporter_fn(status);
-                  exporter_initialized_ = true;
-                  RAY_LOG(INFO) << "Exporter initialized.";
-                }
-              });
-  if (retry_count >= max_retry) {
-    init_exporter_fn(
-        Status::RpcError("Running out of retries to initialize the metrics agent.", 14));
-    return;
-  }
-  retry_count++;
-  retry_timer_->expires_after(std::chrono::milliseconds(retry_interval_ms));
-  retry_timer_->async_wait(
-      [this, init_exporter_fn, retry_count, max_retry, retry_interval_ms](
-          const boost::system::error_code &error) {
-        if (!error) {
-          WaitForServerReadyWithRetry(
-              init_exporter_fn, retry_count, max_retry, retry_interval_ms);
-        } else {
-          RAY_LOG(ERROR) << "Failed to initialize exporter. Data will not be exported to "
-                            "the metrics agent.";
+  HealthCheck(
+      rpc::HealthCheckRequest(),
+      [this, init_exporter_fn, retry_count, max_retry, retry_interval_ms](auto &status,
+                                                                          auto &&reply) {
+        if (status.ok()) {
+          if (exporter_initialized_) {
+            return;
+          }
+          init_exporter_fn(status);
+          exporter_initialized_ = true;
+          RAY_LOG(INFO) << "Exporter initialized.";
+          return;
+        }
+        if (retry_count >= max_retry) {
+          init_exporter_fn(Status::RpcError(
+              "Running out of retries to initialize the metrics agent.", 14));
+          return;
         }
+        io_service_.post(
+            [this, init_exporter_fn, retry_count, max_retry, retry_interval_ms]() {
+              WaitForServerReadyWithRetry(
+                  init_exporter_fn, retry_count + 1, max_retry, retry_interval_ms);
+            },
+            "MetricsAgentClient.WaitForServerReadyWithRetry",
+            retry_interval_ms * 1000);
       });
 }
 

src/ray/rpc/metrics_agent_client.h

@@ -72,12 +72,12 @@ class MetricsAgentClientImpl : public MetricsAgentClient {
   MetricsAgentClientImpl(const std::string &address,
                          const int port,
                          instrumented_io_context &io_service,
-                         rpc::ClientCallManager &client_call_manager) {
+                         rpc::ClientCallManager &client_call_manager)
+      : io_service_(io_service) {
     RAY_LOG(DEBUG) << "Initiate the metrics client of address:"
                    << BuildAddress(address, port);
     grpc_client_ =
         std::make_unique<GrpcClient<ReporterService>>(address, port, client_call_manager);
-    retry_timer_ = std::make_unique<boost::asio::steady_timer>(io_service);
   };
 
   VOID_RPC_CLIENT_METHOD(ReporterService,
@@ -89,7 +89,7 @@ class MetricsAgentClientImpl : public MetricsAgentClient {
   VOID_RPC_CLIENT_METHOD(ReporterService,
                          HealthCheck,
                          grpc_client_,
-                         /*method_timeout_ms*/ -1,
+                         /*method_timeout_ms*/ kMetricAgentInitRetryDelayMs,
                          override)
 
   /// Wait for the server to be ready. Invokes the callback with the final readiness
@@ -99,8 +99,8 @@ class MetricsAgentClientImpl : public MetricsAgentClient {
  private:
   /// The RPC client.
   std::unique_ptr<GrpcClient<ReporterService>> grpc_client_;
-  /// Timer for retrying to initialize the OpenTelemetry exporter.
-  std::unique_ptr<boost::asio::steady_timer> retry_timer_;
+  /// The io context to run the retry loop.
+  instrumented_io_context &io_service_;
   /// Whether the exporter is initialized.
   bool exporter_initialized_ = false;
   /// Wait for the server to be ready with a retry count. Invokes the callback