[core][stats-die/01] kill STATS in rpc component (#57926)

can-anyscale · web-flow · commit 332b67130139 · 2025-10-21T13:41:21.000-07:00
Signed-off-by: Cuong Nguyen &lt;can@anyscale.com&gt;
diff --git a/src/ray/rpc/BUILD.bazel b/src/ray/rpc/BUILD.bazel
@@ -15,6 +15,7 @@ ray_cc_library(
         "//src/ray/core_worker:__pkg__",
     ],
     deps = [
+        ":metrics",
         ":rpc_callback_types",
         "//src/ray/common:asio",
         "//src/ray/common:grpc_util",
@@ -98,6 +99,7 @@ ray_cc_library(
     hdrs = ["server_call.h"],
     visibility = ["//visibility:private"],
     deps = [
+        ":metrics",
         ":rpc_callback_types",
         "//src/ray/common:asio",
         "//src/ray/common:grpc_util",
@@ -175,3 +177,11 @@ ray_cc_library(
         "//src/ray/common:status",
     ],
 )
+
+ray_cc_library(
+    name = "metrics",
+    hdrs = ["metrics.h"],
+    deps = [
+        "//src/ray/stats:stats_metric",
+    ],
+)
diff --git a/src/ray/rpc/client_call.h b/src/ray/rpc/client_call.h
@@ -30,8 +30,8 @@
 #include "ray/common/grpc_util.h"
 #include "ray/common/id.h"
 #include "ray/common/status.h"
+#include "ray/rpc/metrics.h"
 #include "ray/rpc/rpc_callback_types.h"
-#include "ray/stats/metric_defs.h"
 #include "ray/util/thread_utils.h"
 
 namespace ray {
@@ -104,7 +104,8 @@ class ClientCallImpl : public ClientCall {
       status = return_status_;
     }
     if (record_stats_ && !status.ok()) {
-      stats::STATS_grpc_client_req_failed.Record(1.0, stats_handle_->event_name);
+      grpc_client_req_failed_counter_.Record(1.0,
+                                             {{"Method", stats_handle_->event_name}});
     }
     if (callback_ != nullptr) {
       // This should be only called once.
@@ -146,6 +147,9 @@ class ClientCallImpl : public ClientCall {
   /// the server and/or tweak certain RPC behaviors.
   grpc::ClientContext context_;
 
+  ray::stats::Count grpc_client_req_failed_counter_{
+      GetGrpcClientReqFailedCounterMetric()};
+
   friend class ClientCallManager;
 };
 
diff --git a/src/ray/rpc/metrics.h b/src/ray/rpc/metrics.h
@@ -0,0 +1,80 @@
+// 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 {
+namespace rpc {
+
+inline ray::stats::Histogram GetGrpcServerReqProcessTimeMsHistogramMetric() {
+  return ray::stats::Histogram(
+      /*name=*/"grpc_server_req_process_time_ms",
+      /*description=*/"Request latency in grpc server",
+      /*unit=*/"",
+      /*boundaries=*/{0.1, 1, 10, 100, 1000, 10000},
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcServerReqNewCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_server_req_new",
+      /*description=*/"New request number in grpc server",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcServerReqHandlingCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_server_req_handling",
+      /*description=*/"Request number are handling in grpc server",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcServerReqFinishedCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_server_req_finished",
+      /*description=*/"Finished request number in grpc server",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcServerReqSucceededCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_server_req_succeeded",
+      /*description=*/"Succeeded request count in grpc server",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcServerReqFailedCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_server_req_failed",
+      /*description=*/"Failed request count in grpc server",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+inline ray::stats::Count GetGrpcClientReqFailedCounterMetric() {
+  return ray::stats::Count(
+      /*name=*/"grpc_client_req_failed",
+      /*description=*/"Number of gRPC client failures (non-OK response statuses).",
+      /*unit=*/"",
+      /*tag_keys=*/{"Method"});
+}
+
+}  // namespace rpc
+}  // namespace ray
diff --git a/src/ray/rpc/server_call.h b/src/ray/rpc/server_call.h
@@ -27,9 +27,9 @@
 #include "ray/common/grpc_util.h"
 #include "ray/common/id.h"
 #include "ray/common/status.h"
+#include "ray/rpc/metrics.h"
 #include "ray/rpc/rpc_callback_types.h"
 #include "ray/stats/metric.h"
-#include "ray/stats/metric_defs.h"
 
 namespace ray {
 namespace rpc {
@@ -185,7 +185,7 @@ class ServerCallImpl : public ServerCall {
     // TODO(Yi Cheng) call_name_ sometimes get corrunpted due to memory issues.
     RAY_CHECK(!call_name_.empty()) << "Call name is empty";
     if (record_metrics_) {
-      ray::stats::STATS_grpc_server_req_new.Record(1.0, call_name_);
+      grpc_server_req_new_counter_.Record(1.0, {{"Method", call_name_}});
     }
   }
 
@@ -218,7 +218,7 @@ class ServerCallImpl : public ServerCall {
 
     start_time_ = absl::GetCurrentTimeNanos();
     if (record_metrics_) {
-      ray::stats::STATS_grpc_server_req_handling.Record(1.0, call_name_);
+      grpc_server_req_handling_counter_.Record(1.0, {{"Method", call_name_}});
     }
     if (!io_service_.stopped()) {
       io_service_.post([this, auth_success] { HandleRequestImpl(auth_success); },
@@ -279,8 +279,8 @@ class ServerCallImpl : public ServerCall {
 
   void OnReplySent() override {
     if (record_metrics_) {
-      ray::stats::STATS_grpc_server_req_finished.Record(1.0, call_name_);
-      ray::stats::STATS_grpc_server_req_succeeded.Record(1.0, call_name_);
+      grpc_server_req_finished_counter_.Record(1.0, {{"Method", call_name_}});
+      grpc_server_req_succeeded_counter_.Record(1.0, {{"Method", call_name_}});
     }
     if (send_reply_success_callback_ && !io_service_.stopped()) {
       io_service_.post(
@@ -292,8 +292,8 @@ class ServerCallImpl : public ServerCall {
 
   void OnReplyFailed() override {
     if (record_metrics_) {
-      ray::stats::STATS_grpc_server_req_finished.Record(1.0, call_name_);
-      ray::stats::STATS_grpc_server_req_failed.Record(1.0, call_name_);
+      grpc_server_req_finished_counter_.Record(1.0, {{"Method", call_name_}});
+      grpc_server_req_failed_counter_.Record(1.0, {{"Method", call_name_}});
     }
     if (send_reply_failure_callback_ && !io_service_.stopped()) {
       io_service_.post(
@@ -311,8 +311,8 @@ class ServerCallImpl : public ServerCall {
     EventTracker::RecordEnd(std::move(stats_handle_));
     auto end_time = absl::GetCurrentTimeNanos();
     if (record_metrics_) {
-      ray::stats::STATS_grpc_server_req_process_time_ms.Record(
-          (end_time - start_time_) / 1000000.0, call_name_);
+      grpc_server_req_process_time_ms_histogram_.Record(
+          (end_time - start_time_) / 1000000.0, {{"Method", call_name_}});
     }
   }
 
@@ -385,6 +385,18 @@ class ServerCallImpl : public ServerCall {
   /// If true, the server call will generate gRPC server metrics.
   bool record_metrics_;
 
+  ray::stats::Histogram grpc_server_req_process_time_ms_histogram_{
+      GetGrpcServerReqProcessTimeMsHistogramMetric()};
+  ray::stats::Count grpc_server_req_new_counter_{GetGrpcServerReqNewCounterMetric()};
+  ray::stats::Count grpc_server_req_handling_counter_{
+      GetGrpcServerReqHandlingCounterMetric()};
+  ray::stats::Count grpc_server_req_finished_counter_{
+      GetGrpcServerReqFinishedCounterMetric()};
+  ray::stats::Count grpc_server_req_succeeded_counter_{
+      GetGrpcServerReqSucceededCounterMetric()};
+  ray::stats::Count grpc_server_req_failed_counter_{
+      GetGrpcServerReqFailedCounterMetric()};
+
   template <class T1, class T2, class T3, class T4, AuthType T5>
   friend class ServerCallFactoryImpl;
 };
diff --git a/src/ray/stats/metric.cc b/src/ray/stats/metric.cc
@@ -181,8 +181,6 @@ void Metric::RecordForCython(double value,
   Record(value, std::move(tags_pair_vec));
 }
 
-Metric::~Metric() { opencensus::stats::StatsExporter::RemoveView(name_); }
-
 void Gauge::RegisterOpenTelemetryMetric() {
   // Register the metric in OpenTelemetry.
   OpenTelemetryMetricRecorder::GetInstance().RegisterGaugeMetric(name_, description_);
diff --git a/src/ray/stats/metric.h b/src/ray/stats/metric.h
@@ -108,7 +108,7 @@ class Metric : public observability::MetricInterface {
          std::string unit,
          const std::vector<std::string> &tag_keys = {});
 
-  ~Metric() override;
+  ~Metric() = default;
 
   Metric &operator()() { return *this; }
 
diff --git a/src/ray/stats/metric_defs.cc b/src/ray/stats/metric_defs.cc
@@ -110,46 +110,6 @@ DEFINE_stats(operation_queue_time_ms,
 DEFINE_stats(
     operation_active_count, "active operation number", ("Name"), (), ray::stats::GAUGE);
 
-/// GRPC server
-DEFINE_stats(grpc_server_req_process_time_ms,
-             "Request latency in grpc server",
-             ("Method"),
-             ({0.1, 1, 10, 100, 1000, 10000}, ),
-             ray::stats::HISTOGRAM);
-DEFINE_stats(grpc_server_req_new,
-             "New request number in grpc server",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-DEFINE_stats(grpc_server_req_handling,
-             "Request number are handling in grpc server",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-DEFINE_stats(grpc_server_req_finished,
-             "Finished request number in grpc server",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-DEFINE_stats(grpc_server_req_succeeded,
-             "Succeeded request count in grpc server",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-DEFINE_stats(grpc_server_req_failed,
-             "Failed request count in grpc server",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-
-/// Number of failures observed from gRPC client(s).
-/// A failure is an RPC whose response status was not `OK`.
-DEFINE_stats(grpc_client_req_failed,
-             "Number of gRPC client failures (non-OK response statuses).",
-             ("Method"),
-             (),
-             ray::stats::COUNT);
-
 /// Object Manager.
 DEFINE_stats(object_manager_bytes,
              "Number of bytes pushed or received by type {PushedFromLocalPlasma, "
diff --git a/src/ray/stats/metric_defs.h b/src/ray/stats/metric_defs.h
@@ -51,17 +51,6 @@ DECLARE_stats(operation_run_time_ms);
 DECLARE_stats(operation_queue_time_ms);
 DECLARE_stats(operation_active_count);
 
-/// GRPC server
-DECLARE_stats(grpc_server_req_process_time_ms);
-DECLARE_stats(grpc_server_req_new);
-DECLARE_stats(grpc_server_req_handling);
-DECLARE_stats(grpc_server_req_finished);
-DECLARE_stats(grpc_server_req_succeeded);
-DECLARE_stats(grpc_server_req_failed);
-
-/// GRPC Client Failures
-DECLARE_stats(grpc_client_req_failed);
-
 /// Object Manager.
 DECLARE_stats(object_manager_bytes);
 DECLARE_stats(object_manager_received_chunks);

Original file line number	Diff line number	Diff line change
`@@ -181,8 +181,6 @@ void Metric::RecordForCython(double value,`
`181`	`181`	`Record(value, std::move(tags_pair_vec));`
`182`	`182`	`}`
`183`	`183`
`184`		`-Metric::~Metric() { opencensus::stats::StatsExporter::RemoveView(name_); }`
`185`		`-`
`186`	`184`	`void Gauge::RegisterOpenTelemetryMetric() {`
`187`	`185`	`// Register the metric in OpenTelemetry.`
`188`	`186`	`OpenTelemetryMetricRecorder::GetInstance().RegisterGaugeMetric(name_, description_);`