[core] Fix "RayEventRecorder::StartExportingEvents() should be called only once." (#57917)

This PR fixes the Ray check failure
RayEventRecorder::StartExportingEvents() should be called only once..
The failure can occur in the following scenario:
- The metric_agent_client successfully establishes a connection with the
dashboard agent. In this case, RayEventRecorder::StartExportingEvents is
correctly invoked to start sending events.
- At the same time, the metric_agent_client exceeds its maximum number
of connection retries. In this case,
RayEventRecorder::StartExportingEvents is invoked again incorrectly,
causing duplicate attempts to start exporting events.

This PR introduces two fixes:
- In metric_agent_client, the connection success and retry logic are now
synchronized (previously they ran asynchronously, allowing both paths to
trigger).
- Do not call StartExportingEvents if the connection cannot be
established.

Test:
- CI

---------

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

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();
+    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/rpc/metrics_agent_client.cc

@@ -41,30 +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("The metrics agent server is not ready.", 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