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

[can-anyscale]

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

[can-anyscale]

node: we don't perform retry async anymore; instead, we'll retry during the callback of the connection healthcheck

[can-anyscale]

set a finite amount of timed out so we don't hang indefinitely in retrying

[jjyao]

Pass in directly when you call the PRC, I think it's more direct

[gemini-code-assist]

Code Review

This pull request addresses a race condition that could lead to a RAY_CHECK failure in RayEventRecorder::StartExportingEvents(). The fix involves two main changes: making StartExportingEvents aware of the connection status, and synchronizing the connection logic in MetricsAgentClientImpl. The changes are logical and move towards resolving the issue. However, I've identified a remaining race condition in MetricsAgentClientImpl on the exporter_initialized_ flag, which could still lead to the same RAY_CHECK failure under concurrent execution. My review includes a detailed comment and suggestions to make the implementation fully thread-safe.

[gemini-code-assist]

This block has a race condition on exporter_initialized_. Since the io_service can use multiple threads, two concurrent HealthCheck callbacks could both see exporter_initialized_ as false, leading to init_exporter_fn being called twice. This would re-introduce the RAY_CHECK failure this PR aims to fix.

To solve this, you should protect this critical section. One way is to use a mutex. You would add absl::Mutex exporter_mutex_; to MetricsAgentClientImpl (in the .h file, and include absl/synchronization/mutex.h) and then use it here:

if (status.ok()) {
  absl::MutexLock lock(&exporter_mutex_);
  if (exporter_initialized_) {
    return;
  }
  init_exporter_fn(status);
  exporter_initialized_ = true;
  RAY_LOG(INFO) << "Exporter initialized.";
  return;
}

An alternative is to make exporter_initialized_ an std::atomic<bool> (which would require including <atomic>).

[can-anyscale]

thanks, io_context is single threaded

[edoakes]

this is a surprising pattern -- why not just avoid calling StartExportingEvents if the health check fails?

[can-anyscale]

I think at some point in the past you suggested (or I might have misunderstood) that we should propagate the error status to these sub-components (e.g., otelrecorder, eventrecorder, etc.) so they can handle errors themselve, which is the current pattern here.

But yes, it’s probably better not to call these sub-components at all.

[can-anyscale]

ok, let's just NOT call export then

[can-anyscale]

i'll refactor this call to use the same pattern in another follow up (to make this PR minimal)

[can-anyscale]

@edoakes , @jjyao for re-review, thankks

[edoakes]

build failure @can-anyscale

[can-anyscale]

@edoakes - my bad, fixed now, tested that it builds locally

[cursor]

Bug: Incorrect Retry Timing in Metrics Exporter

The io_service_.post() call uses retry_interval_ms * 1000 for its delay parameter. Since retry_interval_ms is in milliseconds, this multiplication likely causes incorrect retry timing, making retries either too fast or too slow depending on the expected unit. This impacts the metrics exporter's initialization.