[core][gcs] Split out Redis wrapping and capture TableName as std::string

Address review on #65583:

- Move the Redis-backend wrapping (MaybeObserve, the
  gcs_redis_storage_metrics_enabled flag, and the shared_ptr delegate
  change) out of this PR into a separate one.
- NormalizeTableNameLabel now returns a std::string and the latency
  callbacks capture it by value. The string_view-into-static-storage
  design saved at most one SSO copy per operation, which is not
  measurable next to the storage round trip, so drop it along with the
  lifetime/SSO comments and the tests that only existed to support it.

Signed-off-by: zzchun <zzchun8@gmail.com>

Author: zzchun

doc/source/ray-observability/reference/system-metrics.md

@@ -50,7 +50,7 @@ These variables must be set in the Ray node and Dashboard Reporter process envir
   - Current number of placement groups by state. The State label (e.g., PENDING, CREATED, REMOVED) describes the state of the placement group. See [rpc::PlacementGroupTable](https://github.com/ray-project/ray/blob/e85355b9b593742b4f5cb72cab92051980fa73d3/src/ray/protobuf/gcs.proto#L517) for more information.
 * - `ray_gcs_storage_operation_count_total`
   - `Operation`, `TableName`
-  - Number of operations the GCS invoked on its storage backend, broken down by operation (`Put`, `Get`, `GetAll`, `MultiGet`, `Delete`, `BatchDelete`, `GetKeys`, `Exists`, `GetNextJobID`) and by the GCS table it addressed. `TableName` is bounded to the names of Ray's internal `TablePrefix` enum, plus `JobCounter` for `GetNextJobID` and `OTHER` for a table name outside that enum. The values the GCS produces in practice are `JOB`, `ACTOR`, `ACTOR_TASK_SPEC`, `PLACEMENT_GROUP`, `NODE`, `WORKERS` and `KV`; the remaining enum names are legacy and no current code path emits them. Keys, internal-KV namespaces and any other user-supplied string never appear in labels. Recorded when the operation is issued, so this leads `ray_gcs_storage_operation_latency_ms` by the number of operations in flight. On the external-Redis backend, set `RAY_gcs_redis_storage_metrics_enabled=false` to stop exporting both metrics; the in-memory and RocksDB backends always export them.
+  - Number of operations the GCS invoked on its storage backend, broken down by operation (`Put`, `Get`, `GetAll`, `MultiGet`, `Delete`, `BatchDelete`, `GetKeys`, `Exists`, `GetNextJobID`) and by the GCS table it addressed. `TableName` is bounded to the names of Ray's internal `TablePrefix` enum, plus `JobCounter` for `GetNextJobID` and `OTHER` for a table name outside that enum. The values the GCS produces in practice are `JOB`, `ACTOR`, `ACTOR_TASK_SPEC`, `PLACEMENT_GROUP`, `NODE`, `WORKERS` and `KV`; the remaining enum names are legacy and no current code path emits them. Keys, internal-KV namespaces and any other user-supplied string never appear in labels. Recorded when the operation is issued, so this leads `ray_gcs_storage_operation_latency_ms` by the number of operations in flight. Exported only when the GCS storage backend is in-memory or RocksDB.
 * - `ray_gcs_storage_operation_latency_ms`
   - `Operation`, `TableName`
   - Time to invoke an operation on the GCS storage backend, with the same label domain as `ray_gcs_storage_operation_count_total`. Recorded when the operation completes.

src/ray/common/ray_config_def.h

@@ -443,15 +443,6 @@ RAY_CONFIG(int64_t, redis_tcp_keepalive_interval_seconds, 30)
 /// keepalive is enabled.
 RAY_CONFIG(int64_t, redis_tcp_keepalive_probes, 3)
 
-/// Whether the GCS wraps its Redis store client in ObservableStoreClient, which
-/// is what makes gcs_storage_operation_count and
-/// gcs_storage_operation_latency_ms exist on the external-Redis backend at all.
-/// Costs two ray::stats::Metric::Record calls per storage operation on a path
-/// that previously had none. Set to false to restore the pre-change behavior;
-/// the in-memory and RocksDB backends are unaffected either way, because they
-/// have always been wrapped.
-RAY_CONFIG(bool, gcs_redis_storage_metrics_enabled, true)
-
 /// Number of retries for a redis request failure.
 RAY_CONFIG(size_t, num_redis_request_retries, 5)
 

src/ray/gcs/gcs_server.cc

@@ -221,16 +221,7 @@ GcsServer::GcsServer(const ray::gcs::GcsServerConfig &config,
         RayConfig::instance().gcs_redis_heartbeat_interval_milliseconds(),
         "GCSServer.redis_health_check");
 
-    // Observe the Redis backend like every other one. The health check above
-    // keeps its own concrete pointer and so bypasses the wrapper, which is
-    // correct: a PING is not a storage operation and should not appear in
-    // gcs_storage_operation_count. Only the Redis backend is gated -- the other
-    // two have always been observed, so there is nothing new to roll back.
-    store_client = MaybeObserve(std::move(redis_store_client),
-                                RayConfig::instance().gcs_redis_storage_metrics_enabled(),
-                                metrics_.storage_operation_latency_in_ms_histogram,
-                                metrics_.storage_operation_count_counter,
-                                clock_);
+    store_client = redis_store_client;
     break;
   }
 #if defined(__linux__)
@@ -1011,16 +1002,11 @@ void GcsServer::InitUsageStatsClient() {
 
 void GcsServer::InitKVManager() {
   auto &io_context = io_context_provider_.GetIOContext<GcsInternalKVManager>();
-  std::shared_ptr<StoreClient> store_client;
+  std::unique_ptr<StoreClient> store_client;
   switch (storage_type_) {
   case (StorageType::REDIS_PERSIST):
-    store_client = MaybeObserve(
-        std::make_shared<RedisStoreClient>(
-            io_context, GetRedisClientOptions(), clock_, MakeRedisMetrics(metrics_)),
-        RayConfig::instance().gcs_redis_storage_metrics_enabled(),
-        metrics_.storage_operation_latency_in_ms_histogram,
-        metrics_.storage_operation_count_counter,
-        clock_);
+    store_client = std::make_unique<RedisStoreClient>(
+        io_context, GetRedisClientOptions(), clock_, MakeRedisMetrics(metrics_));
     break;
   case (StorageType::IN_MEMORY):
     store_client = std::make_unique<ObservableStoreClient>(

src/ray/gcs/store_client/observable_store_client.cc

@@ -29,17 +29,6 @@ namespace {
 
 // Both tags for one recording site. `table` must already be normalized (see
 // NormalizeTableNameLabel) so that TableName stays inside a bounded domain.
-//
-// MetricInterface::Record takes the tag value as an owned std::string, so the
-// copy here is unavoidable. Every table name a StoreClient caller can actually
-// produce is at most 15 characters (PLACEMENT_GROUP and ACTOR_TASK_SPEC are the
-// longest), which fits libstdc++'s small-string buffer, so in practice it does
-// not allocate. The label domain is wider than that -- NormalizeTableNameLabel
-// accepts every rpc::TablePrefix name, and four of those exceed 15 characters,
-// PLACEMENT_GROUP_SCHEDULE being the longest at 24 -- but none of them is
-// reachable from any current caller, and reaching one would cost a small
-// allocation rather than a wrong label. TableNameLabelTest pins the
-// reachable-name bound so this comment cannot rot silently.
 std::vector<std::pair<std::string_view, std::string>> MetricTags(
     std::string_view operation, std::string_view table) {
   return {{"Operation", std::string(operation)}, {"TableName", std::string(table)}};
@@ -55,35 +44,13 @@ double ElapsedMs(ClockInterface &clock, SteadyTimePoint start) {
 
 }  // namespace
 
-std::shared_ptr<StoreClient> MaybeObserve(
-    std::shared_ptr<StoreClient> delegate,
-    bool enabled,
-    ray::observability::MetricInterface &storage_operation_latency_in_ms_histogram,
-    ray::observability::MetricInterface &storage_operation_count_counter,
-    ClockInterface &clock) {
-  if (!enabled) {
-    return delegate;
-  }
-  return std::make_shared<ObservableStoreClient>(
-      std::move(delegate),
-      storage_operation_latency_in_ms_histogram,
-      storage_operation_count_counter,
-      clock);
-}
-
 void ObservableStoreClient::AsyncPut(const std::string &table_name,
                                      const std::string &key,
                                      std::string data,
                                      bool overwrite,
                                      Postable<void(bool)> callback) {
   const auto start = clock_.SteadyNow();
-  // Normalized once per operation and captured as a view, not as a string: the
-  // observer below runs on the io_context after this call has returned, so a
-  // view borrowed from `table_name` could outlive the caller's string and
-  // dangle. NormalizeTableNameLabel returns a view into protobuf's static
-  // descriptor storage instead, which is why capturing it costs no allocation
-  // and carries no lifetime risk.
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("Put", table));
   delegate_->AsyncPut(table_name,
                       key,
@@ -100,7 +67,7 @@ void ObservableStoreClient::AsyncGet(
     const std::string &key,
     ToPostable<rpc::OptionalItemCallback<std::string>> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("Get", table));
   delegate_->AsyncGet(
       table_name, key, std::move(callback).OnInvocation([this, start, table]() {
@@ -113,7 +80,7 @@ void ObservableStoreClient::AsyncGetAll(
     const std::string &table_name,
     Postable<void(absl::flat_hash_map<std::string, std::string>)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("GetAll", table));
   delegate_->AsyncGetAll(table_name,
                          std::move(callback).OnInvocation([this, start, table]() {
@@ -127,7 +94,7 @@ void ObservableStoreClient::AsyncMultiGet(
     const std::vector<std::string> &keys,
     Postable<void(absl::flat_hash_map<std::string, std::string>)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("MultiGet", table));
   delegate_->AsyncMultiGet(
       table_name, keys, std::move(callback).OnInvocation([this, start, table]() {
@@ -140,7 +107,7 @@ void ObservableStoreClient::AsyncDelete(const std::string &table_name,
                                         const std::string &key,
                                         Postable<void(bool)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("Delete", table));
   delegate_->AsyncDelete(
       table_name, key, std::move(callback).OnInvocation([this, start, table]() {
@@ -153,7 +120,7 @@ void ObservableStoreClient::AsyncBatchDelete(const std::string &table_name,
                                              const std::vector<std::string> &keys,
                                              Postable<void(int64_t)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("BatchDelete", table));
   delegate_->AsyncBatchDelete(
       table_name, keys, std::move(callback).OnInvocation([this, start, table]() {
@@ -182,7 +149,7 @@ void ObservableStoreClient::AsyncGetKeys(
     const std::string &prefix,
     Postable<void(std::vector<std::string>)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("GetKeys", table));
   delegate_->AsyncGetKeys(
       table_name, prefix, std::move(callback).OnInvocation([this, start, table]() {
@@ -195,7 +162,7 @@ void ObservableStoreClient::AsyncExists(const std::string &table_name,
                                         const std::string &key,
                                         Postable<void(bool)> callback) {
   const auto start = clock_.SteadyNow();
-  const std::string_view table = NormalizeTableNameLabel(table_name);
+  const std::string table = NormalizeTableNameLabel(table_name);
   storage_operation_count_counter_.Record(1, MetricTags("Exists", table));
   delegate_->AsyncExists(
       table_name, key, std::move(callback).OnInvocation([this, start, table]() {

src/ray/gcs/store_client/observable_store_client.h

@@ -27,46 +27,11 @@ namespace ray {
 
 namespace gcs {
 
-/// Wraps `delegate` in an ObservableStoreClient when `enabled`, and returns it
-/// unchanged otherwise.
-///
-/// Exists as a named function, and takes `enabled` rather than reading
-/// RayConfig itself, so that both GCS construction sites share one branch and
-/// so that the branch is testable without standing up a GcsServer.
-///
-/// \param delegate The store client to wrap.
-/// \param enabled Whether to observe. GcsServer passes
-/// RayConfig::gcs_redis_storage_metrics_enabled() for the Redis backend; the
-/// in-memory and RocksDB backends are always observed and do not call this.
-/// \param storage_operation_latency_in_ms_histogram Sink for per-operation
-/// latency, recorded when each operation completes.
-/// \param storage_operation_count_counter Sink for per-operation counts,
-/// recorded when each operation is issued.
-/// \param clock Clock used to measure operation latency.
-/// \return Either an ObservableStoreClient owning `delegate`, or `delegate`.
-std::shared_ptr<StoreClient> MaybeObserve(
-    std::shared_ptr<StoreClient> delegate,
-    bool enabled,
-    ray::observability::MetricInterface &storage_operation_latency_in_ms_histogram,
-    ray::observability::MetricInterface &storage_operation_count_counter,
-    ClockInterface &clock);
-
 /// Wraps around a StoreClient instance and observe the metrics.
 class ObservableStoreClient : public StoreClient {
  public:
-  /// \param delegate The store client to observe. Taken as a shared_ptr rather
-  /// than a unique_ptr because the Redis backend has to be reachable
-  /// concretely as well: AsyncCheckHealth is declared on RedisStoreClient and
-  /// not on this interface, so GcsServer keeps its own pointer for the periodic
-  /// health check while handing a copy here. std::unique_ptr converts
-  /// implicitly, so the in-memory and RocksDB construction sites are unchanged.
-  /// \param storage_operation_latency_in_ms_histogram Sink for per-operation
-  /// latency, recorded when each operation completes.
-  /// \param storage_operation_count_counter Sink for per-operation counts,
-  /// recorded when each operation is issued.
-  /// \param clock Clock used to measure operation latency.
   explicit ObservableStoreClient(
-      std::shared_ptr<StoreClient> delegate,
+      std::unique_ptr<StoreClient> delegate,
       ray::observability::MetricInterface &storage_operation_latency_in_ms_histogram,
       ray::observability::MetricInterface &storage_operation_count_counter,
       ClockInterface &clock)
@@ -114,7 +79,7 @@ class ObservableStoreClient : public StoreClient {
                    Postable<void(bool)> callback) override;
 
  private:
-  std::shared_ptr<StoreClient> delegate_;
+  std::unique_ptr<StoreClient> delegate_;
   ray::observability::MetricInterface &storage_operation_latency_in_ms_histogram_;
   ray::observability::MetricInterface &storage_operation_count_counter_;
   ClockInterface &clock_;

src/ray/gcs/store_client/table_name_label.cc

@@ -14,22 +14,18 @@
 
 #include "ray/gcs/store_client/table_name_label.h"
 
-#include <string_view>
+#include <string>
 
 #include "src/ray/protobuf/gcs.pb.h"
 
 namespace ray::gcs {
 
-std::string_view NormalizeTableNameLabel(std::string_view table_name) {
+std::string NormalizeTableNameLabel(const std::string &table_name) {
   rpc::TablePrefix prefix;
   if (!rpc::TablePrefix_Parse(table_name, &prefix)) {
-    return kUnknownTable;
+    return std::string(kUnknownTable);
   }
-  // Deliberately not `table_name`: that view would dangle once the caller's
-  // string goes out of scope, and the latency label is read from a completion
-  // callback. TablePrefix_Name returns a reference into the generated
-  // descriptor's name storage, which lives for the whole process.
-  return rpc::TablePrefix_Name(prefix);
+  return table_name;
 }
 
 }  // namespace ray::gcs

src/ray/gcs/store_client/table_name_label.h

@@ -14,6 +14,7 @@
 
 #pragma once
 
+#include <string>
 #include <string_view>
 
 namespace ray::gcs {
@@ -33,27 +34,14 @@ inline constexpr std::string_view kJobCounterTable = "JobCounter";
 /// Maps a GCS table name onto a closed label domain: the names of the
 /// rpc::TablePrefix enum, plus kUnknownTable.
 ///
-/// Every table name a StoreClient sees today is produced by TablePrefix_Name
-/// at its definition site -- the six GcsTable subclasses in gcs_table_storage.h
-/// and StoreClientInternalKV -- so this normally returns its argument's
-/// spelling unchanged. What it adds is that the domain is bounded by
-/// construction rather than by convention: StoreClient takes an arbitrary
-/// std::string, so without this a future caller could turn a job or actor id
-/// into unbounded label cardinality.
-///
-/// The round trip through the enum is also what makes the result safe to
-/// capture. Parsing proves the name is a TablePrefix constant; re-deriving the
-/// name from the parsed value returns protobuf's generated static string
-/// rather than a view into the caller's argument, so the result outlives the
-/// call and can be held by a completion callback without copying a std::string
-/// into it.
-///
-/// Self-maintaining: adding a TablePrefix value makes it a valid label with no
-/// change here, and the domain can only grow by editing gcs.proto.
+/// Every table name a StoreClient sees today is produced by TablePrefix_Name,
+/// so this normally returns its argument unchanged. What it adds is that the
+/// domain is bounded by construction rather than by convention: StoreClient
+/// takes an arbitrary std::string, so without this a future caller could turn a
+/// job or actor id into unbounded label cardinality.
 ///
 /// \param table_name The table name a StoreClient method was called with.
-/// \return The matching enum name, or kUnknownTable. Points into static
-/// storage, so it outlives the argument.
-std::string_view NormalizeTableNameLabel(std::string_view table_name);
+/// \return The matching enum name, or kUnknownTable.
+std::string NormalizeTableNameLabel(const std::string &table_name);
 
 }  // namespace ray::gcs