[core][metric] Redefine gcs STATS using metric interface (#56201)

This PR is in the series of unifying all metric definition infra. 

This PR migrates all GCS metrics to use the metric interface. It does
that by creating the metric object inside gcs_server and pass them down
as interfaces to sub-components.

Purely refactoring code and repetitive patterns, easier to review than
the number of file changed tells you.

Test:
- CI

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Refactors GCS and core worker to use injected MetricInterface objects
for all metrics, adding new metric helpers and rewiring constructors,
server startup, storage client, and tests accordingly.
> 
> - **Metrics Infrastructure**:
> - Introduce metric helpers in `src/ray/common/metrics.h` and
`src/ray/gcs/metrics.h` (gauges/histograms/counters for actors, jobs,
placement groups, task events, and GCS storage).
> - Replace direct `stats` usage with `MetricInterface` across GCS and
core worker; rename helpers (e.g., `GetTaskMetric` ->
`GetTaskByStateGaugeMetric`, `GetRayEventRecorderDroppedEventsMetric` ->
`GetRayEventRecorderDroppedEventsCounterMetric`).
> - **GCS Server Refactor**:
> - `GcsServer` now constructs/accepts metric instances and passes them
to subcomponents via `Start`/`DoStart` and init methods.
> - `GcsActorManager`, `GcsJobManager`, `GcsPlacementGroupManager`, and
`GcsTaskManager` constructors updated to receive and record via
`MetricInterface`.
> - `ObservableStoreClient` wraps delegate and records storage metrics
via injected interfaces.
> - **Core Worker**:
> - `TaskCounter` and `CoreWorker` updated to use task/actor state
gauges via injected `MetricInterface`.
> - **Tests/Mocks/Build**:
> - Update mocks and tests to use
`FakeGauge`/`FakeCounter`/`FakeHistogram`; validate metric tags/values.
> - Add Bazel targets/deps for new metric headers and fakes; minor BUILD
wiring adjustments.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
bd5ff5a. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

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

Author: can-anyscale

BUILD.bazel

@@ -119,6 +119,7 @@ ray_cc_library(
         ],
     ),
     deps = [
+        "//src/ray/observability:fake_metric",
         "//src/ray/observability:fake_ray_event_recorder",
     ],
 )

src/mock/ray/gcs/gcs_actor_manager.h

@@ -17,6 +17,7 @@
 #include <gmock/gmock.h>
 
 #include "ray/gcs/gcs_actor_manager.h"
+#include "ray/observability/fake_metric.h"
 #include "ray/observability/fake_ray_event_recorder.h"
 
 namespace ray {
@@ -38,7 +39,9 @@ class MockGcsActorManager : public GcsActorManager {
             [](const ActorID &) {},
             worker_client_pool,
             /*ray_event_recorder=*/fake_ray_event_recorder_,
-            /*session_name=*/"") {}
+            /*session_name=*/"",
+            /*actor_by_state_gauge=*/fake_actor_by_state_gauge_,
+            /*gcs_actor_by_state_gauge=*/fake_gcs_actor_by_state_gauge_) {}
 
   MOCK_METHOD(void,
               HandleRegisterActor,
@@ -85,6 +88,8 @@ class MockGcsActorManager : public GcsActorManager {
 
   instrumented_io_context mock_io_context_do_not_use_;
   observability::FakeRayEventRecorder fake_ray_event_recorder_;
+  observability::FakeGauge fake_actor_by_state_gauge_;
+  observability::FakeGauge fake_gcs_actor_by_state_gauge_;
 };
 
 }  // namespace gcs

src/mock/ray/gcs/gcs_placement_group_manager.h

@@ -17,14 +17,27 @@
 #include <gmock/gmock.h>
 
 #include "ray/gcs/gcs_placement_group_manager.h"
+#include "ray/observability/fake_metric.h"
 
 namespace ray {
 namespace gcs {
 
 class MockGcsPlacementGroupManager : public GcsPlacementGroupManager {
  public:
-  explicit MockGcsPlacementGroupManager(GcsResourceManager &gcs_resource_manager)
-      : GcsPlacementGroupManager(context_, gcs_resource_manager) {}
+  explicit MockGcsPlacementGroupManager(
+      GcsResourceManager &gcs_resource_manager,
+      ray::observability::MetricInterface &placement_group_gauge,
+      ray::observability::MetricInterface
+          &placement_group_creation_latency_in_ms_histogram,
+      ray::observability::MetricInterface
+          &placement_group_scheduling_latency_in_ms_histogram,
+      ray::observability::MetricInterface &placement_group_count_gauge)
+      : GcsPlacementGroupManager(context_,
+                                 gcs_resource_manager,
+                                 placement_group_gauge,
+                                 placement_group_creation_latency_in_ms_histogram,
+                                 placement_group_scheduling_latency_in_ms_histogram,
+                                 placement_group_count_gauge) {}
   MOCK_METHOD(void,
               HandleCreatePlacementGroup,
               (rpc::CreatePlacementGroupRequest request,

src/ray/common/BUILD.bazel

@@ -420,3 +420,11 @@ ray_cc_library(
         "//src/ray/common:status",
     ],
 )
+
+ray_cc_library(
+    name = "metrics",
+    hdrs = ["metrics.h"],
+    deps = [
+        "//src/ray/stats:stats_lib",
+    ],
+)

src/ray/common/metrics.h

@@ -0,0 +1,43 @@
+// Copyright 2025 The Ray Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#pragma once
+
+#include "ray/stats/metric.h"
+
+namespace ray {
+
+inline ray::stats::Gauge GetActorByStateGaugeMetric() {
+  /// Tracks actors by state, including pending, running, and idle actors.
+  ///
+  /// To avoid metric collection conflicts between components reporting on the same actor,
+  /// we use the "Source" required label.
+  return ray::stats::Gauge{
+      /*name=*/"actors",
+      /*description=*/
+      "An actor can be in one of DEPENDENCIES_UNREADY, PENDING_CREATION, ALIVE, "
+      "ALIVE_IDLE, ALIVE_RUNNING_TASKS, RESTARTING, or DEAD states. "
+      "An actor is considered ALIVE_IDLE if it is not executing any tasks.",
+      /*unit=*/"",
+      // State: the actor state, which is from rpc::ActorTableData::ActorState,
+      // For ALIVE actor the sub-state can be IDLE, RUNNING_TASK,
+      // RUNNING_IN_RAY_GET, and RUNNING_IN_RAY_WAIT.
+      // Name: the name of actor class (Keep in sync with the TASK_OR_ACTOR_NAME_TAG_KEY
+      // in python/ray/_private/telemetry/metric_cardinality.py) Source: component
+      // reporting, e.g., "gcs" or "executor".
+      /*tag_keys=*/{"State", "Name", "Source", "JobId"},
+  };
+}
+
+}  // namespace ray

src/ray/core_worker/BUILD.bazel

@@ -32,6 +32,7 @@ ray_cc_library(
         ":reference_counter",
         ":shutdown_coordinator",
         ":task_event_buffer",
+        "//src/ray/common:metrics",
         "//src/ray/common:protobuf_utils",
         "//src/ray/core_worker/task_execution:task_receiver",
         "//src/ray/core_worker/task_submission:normal_task_submitter",

src/ray/core_worker/core_worker.cc

@@ -165,8 +165,10 @@ JobID GetProcessJobID(const CoreWorkerOptions &options) {
   return options.job_id;
 }
 
-TaskCounter::TaskCounter(ray::observability::MetricInterface &task_by_state_counter)
-    : task_by_state_counter_(task_by_state_counter) {
+TaskCounter::TaskCounter(ray::observability::MetricInterface &task_by_state_gauge,
+                         ray::observability::MetricInterface &actor_by_state_gauge)
+    : task_by_state_gauge_(task_by_state_gauge),
+      actor_by_state_gauge_(actor_by_state_gauge) {
   counter_.SetOnChangeCallback(
       [this](const std::tuple<std::string, TaskStatusType, bool>
                  &key) ABSL_EXCLUSIVE_LOCKS_REQUIRED(&mu_) mutable {
@@ -181,31 +183,31 @@ TaskCounter::TaskCounter(ray::observability::MetricInterface &task_by_state_coun
         const auto is_retry_label = is_retry ? "1" : "0";
         // RUNNING_IN_RAY_GET/WAIT are sub-states of RUNNING, so we need to subtract
         // them out to avoid double-counting.
-        task_by_state_counter_.Record(
+        task_by_state_gauge_.Record(
             running_total - num_in_get - num_in_wait,
             {{"State"sv, rpc::TaskStatus_Name(rpc::TaskStatus::RUNNING)},
              {"Name"sv, func_name},
              {"IsRetry"sv, is_retry_label},
              {"JobId"sv, job_id_},
              {"Source"sv, "executor"}});
         // Negate the metrics recorded from the submitter process for these tasks.
-        task_by_state_counter_.Record(
+        task_by_state_gauge_.Record(
             -running_total,
             {{"State"sv, rpc::TaskStatus_Name(rpc::TaskStatus::SUBMITTED_TO_WORKER)},
              {"Name"sv, func_name},
              {"IsRetry"sv, is_retry_label},
              {"JobId"sv, job_id_},
              {"Source"sv, "executor"}});
         // Record sub-state for get.
-        task_by_state_counter_.Record(
+        task_by_state_gauge_.Record(
             num_in_get,
             {{"State"sv, rpc::TaskStatus_Name(rpc::TaskStatus::RUNNING_IN_RAY_GET)},
              {"Name"sv, func_name},
              {"IsRetry"sv, is_retry_label},
              {"JobId"sv, job_id_},
              {"Source"sv, "executor"}});
         // Record sub-state for wait.
-        task_by_state_counter_.Record(
+        task_by_state_gauge_.Record(
             num_in_wait,
             {{"State"sv, rpc::TaskStatus_Name(rpc::TaskStatus::RUNNING_IN_RAY_WAIT)},
              {"Name"sv, func_name},
@@ -226,16 +228,16 @@ void TaskCounter::RecordMetrics() {
     } else {
       running_tasks = 1.0;
     }
-    ray::stats::STATS_actors.Record(idle,
-                                    {{"State", "ALIVE_IDLE"},
-                                     {"Name", actor_name_},
-                                     {"Source", "executor"},
-                                     {"JobId", job_id_}});
-    ray::stats::STATS_actors.Record(running_tasks,
-                                    {{"State", "ALIVE_RUNNING_TASKS"},
-                                     {"Name", actor_name_},
-                                     {"Source", "executor"},
-                                     {"JobId", job_id_}});
+    actor_by_state_gauge_.Record(idle,
+                                 {{"State"sv, "ALIVE_IDLE"},
+                                  {"Name"sv, actor_name_},
+                                  {"Source"sv, "executor"},
+                                  {"JobId"sv, job_id_}});
+    actor_by_state_gauge_.Record(running_tasks,
+                                 {{"State"sv, "ALIVE_RUNNING_TASKS"},
+                                  {"Name"sv, actor_name_},
+                                  {"Source"sv, "executor"},
+                                  {"JobId"sv, job_id_}});
   }
 }
 
@@ -303,7 +305,8 @@ CoreWorker::CoreWorker(
     instrumented_io_context &task_execution_service,
     std::unique_ptr<worker::TaskEventBuffer> task_event_buffer,
     uint32_t pid,
-    ray::observability::MetricInterface &task_by_state_counter)
+    ray::observability::MetricInterface &task_by_state_gauge,
+    ray::observability::MetricInterface &actor_by_state_gauge)
     : options_(std::move(options)),
       get_call_site_(RayConfig::instance().record_ref_creation_sites()
                          ? options_.get_lang_stack
@@ -342,7 +345,7 @@ CoreWorker::CoreWorker(
       task_execution_service_(task_execution_service),
       exiting_detail_(std::nullopt),
       max_direct_call_object_size_(RayConfig::instance().max_direct_call_object_size()),
-      task_counter_(task_by_state_counter),
+      task_counter_(task_by_state_gauge, actor_by_state_gauge),
       task_event_buffer_(std::move(task_event_buffer)),
       pid_(pid),
       actor_shutdown_callback_(options_.actor_shutdown_callback),

src/ray/core_worker/core_worker.h

@@ -69,7 +69,8 @@ class TaskCounter {
   enum class TaskStatusType { kPending, kRunning, kFinished };
 
  public:
-  explicit TaskCounter(ray::observability::MetricInterface &task_by_state_counter);
+  explicit TaskCounter(ray::observability::MetricInterface &task_by_state_gauge,
+                       ray::observability::MetricInterface &actor_by_state_gauge);
 
   void BecomeActor(const std::string &actor_name) {
     absl::MutexLock l(&mu_);
@@ -134,7 +135,8 @@ class TaskCounter {
   // - Name: the name of the function called
   // - IsRetry: whether the task is a retry
   // - Source: component reporting, e.g., "core_worker", "executor", or "pull_manager"
-  ray::observability::MetricInterface &task_by_state_counter_;
+  ray::observability::MetricInterface &task_by_state_gauge_;
+  ray::observability::MetricInterface &actor_by_state_gauge_;
 };
 
 struct TaskToRetry {
@@ -199,7 +201,8 @@ class CoreWorker {
              instrumented_io_context &task_execution_service,
              std::unique_ptr<worker::TaskEventBuffer> task_event_buffer,
              uint32_t pid,
-             ray::observability::MetricInterface &task_by_state_counter);
+             ray::observability::MetricInterface &task_by_state_counter,
+             ray::observability::MetricInterface &actor_by_state_counter);
 
   CoreWorker(CoreWorker const &) = delete;
 

src/ray/core_worker/core_worker_process.cc

@@ -470,7 +470,7 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
         return core_worker->core_worker_client_pool_->GetOrConnect(*addr);
       },
       gcs_client,
-      task_by_state_counter_,
+      task_by_state_gauge_,
       /*free_actor_object_callback=*/
       [this](const ObjectID &object_id) {
         auto core_worker = GetCoreWorker();
@@ -681,7 +681,8 @@ std::shared_ptr<CoreWorker> CoreWorkerProcessImpl::CreateCoreWorker(
                                    task_execution_service_,
                                    std::move(task_event_buffer),
                                    pid,
-                                   task_by_state_counter_);
+                                   task_by_state_gauge_,
+                                   actor_by_state_gauge_);
   return core_worker;
 }
 

src/ray/core_worker/core_worker_process.h

@@ -18,6 +18,7 @@
 #include <memory>
 #include <string>
 
+#include "ray/common/metrics.h"
 #include "ray/core_worker/core_worker_options.h"
 #include "ray/core_worker/grpc_service.h"
 #include "ray/core_worker/metrics.h"
@@ -184,7 +185,8 @@ class CoreWorkerProcessImpl {
   /// The client to export metrics to the metrics agent.
   std::unique_ptr<ray::rpc::MetricsAgentClient> metrics_agent_client_;
 
-  ray::stats::Gauge task_by_state_counter_{GetTaskMetric()};
+  ray::stats::Gauge task_by_state_gauge_{GetTaskByStateGaugeMetric()};
+  ray::stats::Gauge actor_by_state_gauge_{GetActorByStateGaugeMetric()};
 };
 }  // namespace core
 }  // namespace ray