diff --git a/doc/source/ray-observability/reference/system-metrics.rst b/doc/source/ray-observability/reference/system-metrics.rst index c19c43501cdd..8d95f0ea9c34 100644 --- a/doc/source/ray-observability/reference/system-metrics.rst +++ b/doc/source/ray-observability/reference/system-metrics.rst @@ -42,6 +42,15 @@ Ray exports a number of system metrics, which provide introspection into the sta * - `ray_placement_groups` - `State` - 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 `_ for more information. + * - `ray_gcs_redis_request_payload_bytes` + - `Command`, `TableName` + - Application bytes in Redis command arguments sent by the GCS, by command and GCS table. Exported only when the GCS storage backend is Redis. Includes the verb, Redis key, field names, and values; excludes RESP framing, TLS, and TCP/IP overhead. Counted once per logical command, not per retry. `Command` is the uppercase Redis verb truncated to 16 bytes; `TableName` is a GCS table, `NONE`, or `ALL`. + * - `ray_gcs_redis_response_payload_bytes` + - `Command`, `TableName` + - Application bytes in Redis replies received by the GCS, by command and GCS table. Exported only when the GCS storage backend is Redis. Includes bulk/status strings, HSCAN field names, and integer decimal text; excludes nil replies, error replies that are retried, RESP framing, TLS, and TCP/IP overhead. + * - `ray_gcs_redis_command_count_total` + - `Command`, `TableName` + - Number of Redis commands issued by the GCS, by command and GCS table. Exported only when the GCS storage backend is Redis. Counts Redis commands, not `StoreClient` calls: batches count per chunk and table scans per HSCAN round, and a retried command counts once. Divide the byte metrics by this for mean bytes per Redis command. Set `RAY_gcs_redis_payload_metrics_enabled=false` to stop recording all three metrics. The byte metrics omit `_total` under OpenTelemetry because they are non-monotonic sums; legacy OpenCensus also exports them with `_total`. * - `ray_memory_manager_worker_eviction_total` - `Type`, `Name` - The number of tasks and actors killed by the Ray Out of Memory killer (https://docs.ray.io/en/master/ray-core/scheduling/ray-oom-prevention.html) broken down by types (whether it is tasks or actors) and names (name of tasks and actors). diff --git a/src/ray/common/ray_config_def.h b/src/ray/common/ray_config_def.h index 1886afabd028..f0a7203bee3b 100644 --- a/src/ray/common/ray_config_def.h +++ b/src/ray/common/ray_config_def.h @@ -391,6 +391,14 @@ RAY_CONFIG(int64_t, redis_db_connect_wait_milliseconds, 500) /// Timeout for synchronous Redis probe commands issued while initializing GCS storage. RAY_CONFIG(int64_t, redis_db_probe_timeout_milliseconds, 30000) +/// Whether the GCS records per-command Redis payload byte metrics +/// (gcs_redis_request_payload_bytes, gcs_redis_response_payload_bytes, +/// gcs_redis_command_count). Recording costs three additional +/// ray::stats::Metric::Record calls per Redis command, each of which takes a +/// process-global registration mutex. Set to false to restore the pre-change +/// behavior if that contention is measurable in a high-throughput GCS. +RAY_CONFIG(bool, gcs_redis_payload_metrics_enabled, true) + /// Whether GCS namespace cleanup deletes Redis keys with UNLINK instead of DEL. /// This is disabled by default. With Redis's default lazyfree-lazy-user-del=no, /// DEL reclaims memory synchronously; Redis operators can configure DEL itself diff --git a/src/ray/gcs/gcs_server.cc b/src/ray/gcs/gcs_server.cc index bf6fb79f27e1..3e2080819a68 100644 --- a/src/ray/gcs/gcs_server.cc +++ b/src/ray/gcs/gcs_server.cc @@ -50,6 +50,19 @@ namespace ray { namespace gcs { +namespace { + +// The subset of GcsServerMetrics the Redis store client records into. Returned +// by value: RedisMetrics holds references, and the referents are owned by the +// GcsServerMetrics the server was constructed with, which outlives it. +RedisMetrics MakeRedisMetrics(const GcsServerMetrics &metrics) { + return RedisMetrics{metrics.redis_request_payload_bytes_sum, + metrics.redis_response_payload_bytes_sum, + metrics.redis_command_count_counter}; +} + +} // namespace + inline std::ostream &operator<<(std::ostream &str, GcsServer::StorageType val) { switch (val) { case GcsServer::StorageType::IN_MEMORY: @@ -190,8 +203,8 @@ GcsServer::GcsServer(const ray::gcs::GcsServerConfig &config, clock_); break; case StorageType::REDIS_PERSIST: { - auto redis_store_client = - std::make_shared(io_context, GetRedisClientOptions(), clock_); + auto redis_store_client = std::make_shared( + io_context, GetRedisClientOptions(), clock_, MakeRedisMetrics(metrics_)); // Health check Redis periodically and crash if it becomes unavailable. // NOTE: periodical_runner_ must run on the same IO context as the Redis client. periodical_runner_->RunFnPeriodically( @@ -770,8 +783,8 @@ void GcsServer::InitKVManager() { std::unique_ptr store_client; switch (storage_type_) { case (StorageType::REDIS_PERSIST): - store_client = - std::make_unique(io_context, GetRedisClientOptions(), clock_); + store_client = std::make_unique( + io_context, GetRedisClientOptions(), clock_, MakeRedisMetrics(metrics_)); break; case (StorageType::IN_MEMORY): store_client = std::make_unique( diff --git a/src/ray/gcs/gcs_server_main.cc b/src/ray/gcs/gcs_server_main.cc index c462798d3c52..0aba2dceb8e9 100644 --- a/src/ray/gcs/gcs_server_main.cc +++ b/src/ray/gcs/gcs_server_main.cc @@ -200,6 +200,11 @@ int main(int argc, char *argv[]) { ray::gcs::GetGcsStorageOperationLatencyInMsHistogramMetric(); auto storage_operation_count_counter = ray::gcs::GetGcsStorageOperationCountCounterMetric(); + auto redis_request_payload_bytes_sum = + ray::gcs::GetGcsRedisRequestPayloadBytesSumMetric(); + auto redis_response_payload_bytes_sum = + ray::gcs::GetGcsRedisResponsePayloadBytesSumMetric(); + auto redis_command_count_counter = ray::gcs::GetGcsRedisCommandCountCounterMetric(); auto resource_usage_gauge = ray::raylet::GetResourceUsageGaugeMetric(); auto health_check_rpc_latency_ms_histogram = ray::gcs::GetHealthCheckRpcLatencyMsHistogramMetric(); @@ -230,6 +235,9 @@ int main(int argc, char *argv[]) { /*storage_operation_latency_in_ms_histogram=*/ storage_operation_latency_in_ms_histogram, /*storage_operation_count_counter=*/storage_operation_count_counter, + /*redis_request_payload_bytes_sum=*/redis_request_payload_bytes_sum, + /*redis_response_payload_bytes_sum=*/redis_response_payload_bytes_sum, + /*redis_command_count_counter=*/redis_command_count_counter, resource_usage_gauge, scheduler_placement_time_ms_histogram, health_check_rpc_latency_ms_histogram, diff --git a/src/ray/gcs/metrics.h b/src/ray/gcs/metrics.h index 03c66ef39656..ff0b6335074e 100644 --- a/src/ray/gcs/metrics.h +++ b/src/ray/gcs/metrics.h @@ -36,6 +36,9 @@ struct GcsServerMetrics { ray::observability::MetricInterface &event_recorder_dropped_events_counter; ray::observability::MetricInterface &storage_operation_latency_in_ms_histogram; ray::observability::MetricInterface &storage_operation_count_counter; + ray::observability::MetricInterface &redis_request_payload_bytes_sum; + ray::observability::MetricInterface &redis_response_payload_bytes_sum; + ray::observability::MetricInterface &redis_command_count_counter; ray::observability::MetricInterface &resource_usage_gauge; ray::observability::MetricInterface &scheduler_placement_time_ms_histogram; ray::observability::MetricInterface &health_check_rpc_latency_ms_histogram; @@ -168,6 +171,58 @@ inline ray::stats::Count GetGcsStorageOperationCountCounterMetric() { }; } +// The payload definitions below are normative: they are what the metric means, +// and tests assert exact deltas against them. "Payload" is always the +// application data carried by RESP values, and never the RESP framing +// ("*N\r\n", "$len\r\n", type prefixes, trailing CRLF), TLS records or TCP/IP +// headers. +// GCS values are not compressed anywhere between GcsTable::Put and the socket, +// so compressed bytes do not arise. To reconcile with Redis' own +// total_net_input_bytes, add framing: 3 + digits(nargs) per command plus +// 5 + digits(len) per argument. + +inline ray::stats::Sum GetGcsRedisRequestPayloadBytesSumMetric() { + return ray::stats::Sum{ + /*name=*/"gcs_redis_request_payload_bytes", + /*description=*/ + "Bytes of Redis command arguments sent by the GCS: the sum of the byte " + "lengths of the RESP arguments, including the command verb, the Redis " + "key, hash field names and values. Excludes RESP framing, TLS and TCP/IP " + "overhead; GCS values are not compressed. Counted once per logical " + "command, not per retry.", + /*unit=*/"bytes", + /*tag_keys=*/{"Command", "TableName"}, + }; +} + +inline ray::stats::Sum GetGcsRedisResponsePayloadBytesSumMetric() { + return ray::stats::Sum{ + /*name=*/"gcs_redis_response_payload_bytes", + /*description=*/ + "Bytes of Redis replies received by the GCS: the sum of the byte lengths " + "of every bulk string and status string in the reply, including the field " + "names returned by HSCAN. Integers count as their decimal text; nil replies " + "contribute zero. Excludes RESP framing, TLS and TCP/IP overhead. A reply " + "that comes back as an error is retried instead of delivered, and " + "contributes nothing at all -- error bytes are never counted here.", + /*unit=*/"bytes", + /*tag_keys=*/{"Command", "TableName"}, + }; +} + +inline ray::stats::Count GetGcsRedisCommandCountCounterMetric() { + return ray::stats::Count{ + /*name=*/"gcs_redis_command_count", + /*description=*/ + "Number of Redis commands issued by the GCS, broken down by command and " + "table. Batched operations count once per chunk and a table scan counts " + "once per HSCAN round, so this is a count of round trips rather than of " + "StoreClient calls.", + /*unit=*/"", + /*tag_keys=*/{"Command", "TableName"}, + }; +} + inline ray::stats::Histogram GetHealthCheckRpcLatencyMsHistogramMetric() { return ray::stats::Histogram{ /*name=*/"health_check_rpc_latency_ms", diff --git a/src/ray/gcs/store_client/BUILD.bazel b/src/ray/gcs/store_client/BUILD.bazel index 73654b92c509..e4904b4e52c6 100644 --- a/src/ray/gcs/store_client/BUILD.bazel +++ b/src/ray/gcs/store_client/BUILD.bazel @@ -38,6 +38,7 @@ ray_cc_library( "//src/ray/util:network_util", "@boost//:asio", "@com_google_absl//absl/container:flat_hash_set", + "@com_google_absl//absl/strings", "@com_google_absl//absl/strings:str_format", "@com_google_absl//absl/synchronization", ], diff --git a/src/ray/gcs/store_client/redis_context.cc b/src/ray/gcs/store_client/redis_context.cc index aa6e23111310..4f395064e481 100644 --- a/src/ray/gcs/store_client/redis_context.cc +++ b/src/ray/gcs/store_client/redis_context.cc @@ -15,9 +15,13 @@ #include "ray/gcs/store_client/redis_context.h" #include +#include +#include +#include #include #include #include +#include #include #include @@ -30,6 +34,7 @@ extern "C" { } // TODO(pcm): Integrate into the C++ tree. +#include "absl/strings/ascii.h" #include "absl/strings/str_cat.h" #include "absl/strings/str_join.h" #include "absl/strings/str_split.h" @@ -40,6 +45,55 @@ namespace ray { namespace gcs { +size_t ResponsePayloadBytes(const redisReply &reply) { + switch (reply.type) { + case REDIS_REPLY_STRING: + case REDIS_REPLY_STATUS: + // Only reachable for an error nested inside an aggregate reply; a top-level + // error is retried before it gets here. See the header. + case REDIS_REPLY_ERROR: + case REDIS_REPLY_VERB: + case REDIS_REPLY_BIGNUM: + return static_cast(reply.len); + case REDIS_REPLY_INTEGER: { + char buffer[std::numeric_limits::digits10 + 3]; + const auto [end, error] = + std::to_chars(buffer, buffer + sizeof(buffer), reply.integer); + RAY_CHECK(error == std::errc{}); + return static_cast(end - buffer); + } + case REDIS_REPLY_DOUBLE: + // hiredis preserves the original RESP3 decimal representation in `str`. + return static_cast(reply.len); + case REDIS_REPLY_BOOL: + // RESP3 encodes booleans as one application byte: `t` or `f`. + return 1; + case REDIS_REPLY_NIL: + return 0; + case REDIS_REPLY_ARRAY: + case REDIS_REPLY_MAP: + case REDIS_REPLY_SET: + case REDIS_REPLY_PUSH: { + // hiredis leaves `element` uninitialized when `elements` is 0, so the loop + // bound is what keeps this from dereferencing garbage. + size_t total = 0; + for (size_t i = 0; i < reply.elements; ++i) { + total += ResponsePayloadBytes(*reply.element[i]); + } + return total; + } + default: + // Unknown reply types have no payload definition. + return 0; + } +} + +std::string NormalizeRedisCommandLabel(std::string_view verb) { + std::string label(verb.substr(0, kMaxRedisCommandLabelLength)); + absl::AsciiStrToUpper(&label); + return label; +} + CallbackReply::CallbackReply(const redisReply &redis_reply) : reply_type_(redis_reply.type) { switch (reply_type_) { @@ -163,7 +217,9 @@ RedisRequestContext::RedisRequestContext(instrumented_io_context &io_service, RedisCallback callback, RedisAsyncContext *context, std::vector args, - ClockInterface &clock) + ClockInterface &clock, + RedisMetrics *metrics, + std::string_view table_label) : exp_back_off_(RayConfig::instance().redis_retry_base_ms(), RayConfig::instance().redis_retry_multiplier(), RayConfig::instance().redis_retry_max_ms()), @@ -173,12 +229,32 @@ RedisRequestContext::RedisRequestContext(instrumented_io_context &io_service, callback_(std::move(callback)), start_time_(clock.Now()), redis_cmds_(std::move(args)), - clock_(clock) { + clock_(clock), + metrics_(metrics), + table_label_(table_label) { + RAY_CHECK(!redis_cmds_.empty()); + command_label_ = NormalizeRedisCommandLabel(redis_cmds_.front()); argc_.reserve(redis_cmds_.size()); argv_.reserve(redis_cmds_.size()); + // This context owns the argument buffers now, so their sizes are safe to read + // here no matter what the caller moved into them on the way in (AsyncPut, for + // one, moves the caller's value into the RedisCommand). Summing in the loop + // that already computes argc_ also means the request side costs nothing + // beyond the additions. + size_t request_payload_bytes = 0; for (size_t i = 0; i < redis_cmds_.size(); ++i) { argv_.push_back(redis_cmds_[i].data()); argc_.push_back(redis_cmds_[i].size()); + request_payload_bytes += redis_cmds_[i].size(); + } + if (metrics_ != nullptr) { + // Once per logical command. Run() is what retries, so a retransmission is + // not counted again -- see the metric description. + metrics_->request_payload_bytes_sum.Record( + static_cast(request_payload_bytes), + {{"Command", command_label_}, {"TableName", table_label_}}); + metrics_->command_count_counter.Record( + 1, {{"Command", command_label_}, {"TableName", table_label_}}); } } @@ -204,6 +280,15 @@ void RedisRequestContext::RedisResponseFn(redisAsyncContext *async_context, [request_cxt]() { request_cxt->Run(); }, std::chrono::milliseconds(delay)); } else { + // Measure while hiredis still owns the reply, and before anything is posted + // to the io_service: `request_cxt` is deleted at the end of this branch, so + // every read of it has to happen above the post. + if (request_cxt->metrics_ != nullptr) { + request_cxt->metrics_->response_payload_bytes_sum.Record( + static_cast(ResponsePayloadBytes(*redis_reply)), + {{"Command", request_cxt->command_label_}, + {"TableName", request_cxt->table_label_}}); + } auto reply = std::make_shared(*redis_reply); request_cxt->io_service_.post( [reply, callback = std::move(request_cxt->callback_)]() { @@ -244,9 +329,12 @@ void RedisRequestContext::Run() { return Status::RedisError(REPLY->str); \ } -RedisContext::RedisContext(instrumented_io_context &io_service, ClockInterface &clock) +RedisContext::RedisContext(instrumented_io_context &io_service, + ClockInterface &clock, + std::optional metrics) : io_service_(io_service), clock_(clock), + metrics_(std::move(metrics)), context_(nullptr), ssl_context_(nullptr), redis_db_probe_timeout_milliseconds_( @@ -832,13 +920,18 @@ std::unique_ptr RedisContext::RunArgvSync( } void RedisContext::RunArgvAsync(std::vector args, - RedisCallback redis_callback) { + RedisCallback redis_callback, + std::string_view table_label) { RAY_CHECK(redis_async_context_); - auto request_context = new RedisRequestContext(io_service_, - std::move(redis_callback), - redis_async_context_.get(), - std::move(args), - clock_); + RAY_CHECK(!args.empty()); + auto request_context = + new RedisRequestContext(io_service_, + std::move(redis_callback), + redis_async_context_.get(), + std::move(args), + clock_, + metrics_.has_value() ? &*metrics_ : nullptr, + table_label); // RedisRequestContext is thread safe. request_context->Run(); } diff --git a/src/ray/gcs/store_client/redis_context.h b/src/ray/gcs/store_client/redis_context.h index 5f3e681bf9fe..027a3f896413 100644 --- a/src/ray/gcs/store_client/redis_context.h +++ b/src/ray/gcs/store_client/redis_context.h @@ -16,16 +16,20 @@ #include #include +#include #include #include #include +#include #include +#include #include #include "ray/asio/instrumented_io_context.h" #include "ray/common/status.h" #include "ray/common/status_or.h" #include "ray/gcs/store_client/redis_async_context.h" +#include "ray/observability/metric_interface.h" #include "ray/stats/metric.h" #include "ray/stats/tag_defs.h" #include "ray/util/clock.h" @@ -107,13 +111,69 @@ class CallbackReply { /// operation. using RedisCallback = std::function)>; +/// Payload bytes of a Redis reply, recursing into aggregate replies. String and +/// double nodes contribute their decoded length, integers contribute their +/// decimal text length, booleans contribute 1, and nil nodes contribute 0. RESP +/// framing is never included. +/// +/// Error nodes are measured like any other string, but that only matters for an +/// error nested inside an aggregate reply. A *top-level* error never reaches +/// this function: RedisRequestContext::RedisResponseFn routes it into the retry +/// path before recording, which is why +/// gcs_redis_response_payload_bytes documents error replies as contributing +/// nothing. +/// +/// Callers must invoke this while hiredis still owns the reply. Measuring the +/// raw reply rather than the `CallbackReply` built from it keeps the +/// measurement independent of what that copy chooses to keep, and works for +/// reply shapes `CallbackReply` does not parse. +/// +/// \param reply The reply to measure. Must be a live hiredis reply. +/// \return The payload bytes it carries, 0 for replies that carry none. +size_t ResponsePayloadBytes(const redisReply &reply); + +/// Maximum byte length of the normalized Redis Command label. +inline constexpr size_t kMaxRedisCommandLabelLength = 16; + +/// Normalizes a Redis verb for use as a metric label by converting its first +/// kMaxRedisCommandLabelLength bytes to uppercase ASCII. +/// +/// \param verb The Redis command verb to normalize. +/// \return An owned, normalized label value. +std::string NormalizeRedisCommandLabel(std::string_view verb); + +/// A TableName label value for commands that address no single GCS table. +/// Explicit sentinels rather than "": a blank label value is indistinguishable +/// from a dropped label in Grafana, and the two cases below mean different +/// things. +/// +/// kNoTable: the command has no table at all (PING). +/// kAllTables: the command spans the whole storage namespace (SCAN/DEL/UNLINK +/// in the cleanup path). +inline constexpr std::string_view kNoTable = "NONE"; +inline constexpr std::string_view kAllTables = "ALL"; + +/// Metrics recorded for every async Redis command. Held by reference: the +/// referents are owned by GcsServerMetrics and outlive the RedisContext. +struct RedisMetrics { + ray::observability::MetricInterface &request_payload_bytes_sum; + ray::observability::MetricInterface &response_payload_bytes_sum; + ray::observability::MetricInterface &command_count_counter; +}; + class RedisContext; struct RedisRequestContext { + /// \param metrics Payload metrics to record into, or nullptr to record + /// nothing. Null both when the payload metrics are disabled by config and in + /// the standalone namespace-cleanup process, which has no metrics exporter. + /// \param table_label The GCS table label copied into this request. RedisRequestContext(instrumented_io_context &io_service, RedisCallback callback, RedisAsyncContext *context, std::vector args, - ClockInterface &clock); + ClockInterface &clock, + RedisMetrics *metrics, + std::string_view table_label); static void RedisResponseFn(redisAsyncContext *async_context, void *raw_reply, @@ -134,6 +194,13 @@ struct RedisRequestContext { std::vector argc_; ClockInterface &clock_; + // Nullable; see the constructor docs. + RedisMetrics *metrics_; + // Owned copies: the reply is delivered long after the caller's command and + // table label are gone. + std::string command_label_; + std::string table_label_; + // Ray metrics ray::stats::Histogram ray_metric_gcs_latency_{ "gcs_latency", @@ -145,7 +212,20 @@ struct RedisRequestContext { class RedisContext { public: - explicit RedisContext(instrumented_io_context &io_service, ClockInterface &clock); + /// \param metrics Payload metrics for every command run through this context, + /// or nullopt to record nothing. + /// + /// Held **by value** rather than by pointer on purpose. A RedisContext is + /// shared: RedisStoreClient::RedisScanner copies the shared_ptr and keeps + /// itself alive for the duration of a scan, so the context routinely outlives + /// the RedisStoreClient that created it. Borrowing the metrics from the store + /// client would leave RedisRequestContext dereferencing freed storage on the + /// next HSCAN reply. Owning them here ties their lifetime to the object that + /// RedisRequestContext already depends on outliving it (it holds a raw + /// RedisAsyncContext * into this one). + explicit RedisContext(instrumented_io_context &io_service, + ClockInterface &clock, + std::optional metrics = std::nullopt); ~RedisContext(); @@ -158,12 +238,16 @@ class RedisContext { /// Disconnect from the server. void Disconnect(); - /// Run an arbitrary Redis command without a callback. + /// Run an arbitrary Redis command. /// /// \param args The vector of command args to pass to Redis. /// \param redis_callback The Redis callback function. + /// \param table_label The GCS table label for this command. Deliberately not + /// defaulted: a new command site must decide how it is attributed. The + /// command label is derived from args[0]. void RunArgvAsync(std::vector args, - RedisCallback redis_callback = nullptr); + RedisCallback redis_callback, + std::string_view table_label); redisContext *sync_context() { RAY_CHECK(context_); @@ -195,6 +279,10 @@ class RedisContext { instrumented_io_context &io_service_; ClockInterface &clock_; + // Owned; see the constructor docs. Never reassigned after construction, so + // the pointer handed to each RedisRequestContext stays valid for this + // context's lifetime. + std::optional metrics_; std::unique_ptr context_; redisSSLContext *ssl_context_; diff --git a/src/ray/gcs/store_client/redis_store_client.cc b/src/ray/gcs/store_client/redis_store_client.cc index 58e4d35ca13e..5ab05651fb81 100644 --- a/src/ray/gcs/store_client/redis_store_client.cc +++ b/src/ray/gcs/store_client/redis_store_client.cc @@ -16,6 +16,7 @@ #include #include +#include #include #include #include @@ -121,11 +122,21 @@ void RedisStoreClient::MGetValues( RedisStoreClient::RedisStoreClient(instrumented_io_context &io_service, const RedisClientOptions &options, - ClockInterface &clock) + ClockInterface &clock, + std::optional metrics) : io_service_(io_service), options_(options), external_storage_namespace_(::RayConfig::instance().external_storage_namespace()), - primary_context_(std::make_shared(io_service, clock)) { + // Read the kill switch once here instead of at every recording site: when + // it is off the context holds no metrics and the disabled path costs the + // null check RedisRequestContext already performs. The context owns the + // metrics because it outlives this client whenever a scan is in flight -- + // see the RedisContext constructor docs. + primary_context_(std::make_shared( + io_service, + clock, + ::RayConfig::instance().gcs_redis_payload_metrics_enabled() ? std::move(metrics) + : std::nullopt)) { RAY_CHECK(!options.ip.empty()) << "Redis IP address cannot be empty."; RAY_CHECK_OK(primary_context_->Connect(options.ip, options.port, @@ -295,7 +306,8 @@ void RedisStoreClient::SendRedisCmdWithKeys(std::vector keys, return; } } - // Send the actual request + // Send the actual request. RedisContext derives the Command label from the + // first argument, so only the table attribution is supplied here. primary_context_->RunArgvAsync( command.ToRedisArgs(), [this, @@ -312,7 +324,8 @@ void RedisStoreClient::SendRedisCmdWithKeys(std::vector keys, if (redis_callback) { redis_callback(reply); } - }); + }, + command.redis_key.table_name); }; { @@ -414,7 +427,10 @@ void RedisStoreClient::RedisScanner::Scan() { // releases its self_ref in Scan(). [this, self_ref = self_ref_](const std::shared_ptr &reply) { OnScanCallback(reply); - }); + }, + // One round of a multi-round scan: the counters aggregate every round, so + // gcs_redis_command_count on HSCAN is round trips, not AsyncGetAll calls. + redis_key_.table_name); } void RedisStoreClient::RedisScanner::OnScanCallback( @@ -462,7 +478,8 @@ void RedisStoreClient::AsyncGetNextJobID(Postable callback) { std::move(callback)](const std::shared_ptr &reply) mutable { auto job_id = static_cast(reply->ReadAsInteger()); std::move(callback).Post("GcsStore.GetNextJobID", job_id); - }); + }, + "JobCounter"); } void RedisStoreClient::AsyncGetKeys(const std::string &table_name, @@ -509,7 +526,7 @@ void RedisStoreClient::AsyncCheckHealth(Postable callback) { std::move(callback).Dispatch("RedisStoreClient.AsyncCheckHealth", status); }; - primary_context_->RunArgvAsync({"PING"}, redis_callback); + primary_context_->RunArgvAsync({"PING"}, redis_callback, kNoTable); } // Cleans up all Redis HASHes whose keys carry the given external storage @@ -564,9 +581,15 @@ bool RedisDelKeyPrefixSync(const std::string &host, std::vector cmd{ "SCAN", std::to_string(cursor), "MATCH", match_pattern, "COUNT", scan_count}; std::promise> promise; - context.RunArgvAsync(cmd, [&promise](const std::shared_ptr &reply) { - promise.set_value(reply); - }); + // Labels are supplied even though this process has no metrics exporter + // (the RedisContext above is built without a RedisMetrics), so the call + // site stays honest if that ever changes. + context.RunArgvAsync( + cmd, + [&promise](const std::shared_ptr &reply) { + promise.set_value(reply); + }, + kAllTables); auto reply = promise.get_future().get(); std::vector scan_result; @@ -592,10 +615,12 @@ bool RedisDelKeyPrefixSync(const std::string &host, // handful of keys, so there is nothing to gain from batching. auto del_cmd = std::vector{delete_command, key}; std::promise> prom; - context.RunArgvAsync(del_cmd, - [&prom](const std::shared_ptr &callback_reply) { - prom.set_value(callback_reply); - }); + context.RunArgvAsync( + del_cmd, + [&prom](const std::shared_ptr &callback_reply) { + prom.set_value(callback_reply); + }, + kAllTables); return prom.get_future().get()->ReadAsInteger(); }; size_t num_deleted = 0; diff --git a/src/ray/gcs/store_client/redis_store_client.h b/src/ray/gcs/store_client/redis_store_client.h index 5a1169adf8d2..9d2a3626eb19 100644 --- a/src/ray/gcs/store_client/redis_store_client.h +++ b/src/ray/gcs/store_client/redis_store_client.h @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -130,9 +131,15 @@ class RedisStoreClient : public StoreClient { /// /// \param io_service The event loop for this client. Must be single threaded. /// \param options The options for connecting to Redis. + /// \param metrics Payload byte metrics for the commands this client issues, + /// or nullopt to record none. Ignored (treated as nullopt) when + /// RAY_gcs_redis_payload_metrics_enabled is false, so the config is read once + /// here rather than at every recording site. Ownership moves to the + /// RedisContext, which can outlive this client. explicit RedisStoreClient(instrumented_io_context &io_service, const RedisClientOptions &options, - ClockInterface &clock); + ClockInterface &clock, + std::optional metrics = std::nullopt); void AsyncPut(const std::string &table_name, const std::string &key, diff --git a/src/ray/gcs/store_client/tests/redis_callback_reply_test.cc b/src/ray/gcs/store_client/tests/redis_callback_reply_test.cc index 5351689bbe74..0565123069f7 100644 --- a/src/ray/gcs/store_client/tests/redis_callback_reply_test.cc +++ b/src/ray/gcs/store_client/tests/redis_callback_reply_test.cc @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +#include #include #include @@ -117,4 +118,154 @@ TEST(TestCallbackReply, TestErrorReplyDoesNotCrash) { CallbackReply callback_reply(redis_reply_error); ASSERT_TRUE(callback_reply.IsError()); } + +namespace { + +redisReply MakeStringReply(int type, std::string &backing) { + redisReply reply{}; + reply.type = type; + reply.str = backing.data(); + reply.len = static_cast(backing.size()); + return reply; +} + +redisReply MakeArrayReply(redisReply **elements, size_t count) { + redisReply reply{}; + reply.type = REDIS_REPLY_ARRAY; + reply.elements = count; + reply.element = elements; + return reply; +} + +} // namespace + +// ResponsePayloadBytes is the normative definition of +// gcs_redis_response_payload_bytes: application bytes carried by every reply +// node, recursing into arrays and excluding RESP framing. +// These cases pin that definition; changing any expectation here changes what +// the exported metric means. +TEST(TestResponsePayloadBytes, ScalarReplies) { + { + redisReply nil{}; + nil.type = REDIS_REPLY_NIL; + ASSERT_EQ(ResponsePayloadBytes(nil), 0u); + } + { + // RESP integers carry their ASCII decimal form between ':' and CRLF. + redisReply integer{}; + integer.type = REDIS_REPLY_INTEGER; + integer.integer = 0; + ASSERT_EQ(ResponsePayloadBytes(integer), 1u); + integer.integer = 1; + ASSERT_EQ(ResponsePayloadBytes(integer), 1u); + integer.integer = -12345; + ASSERT_EQ(ResponsePayloadBytes(integer), 6u); + integer.integer = std::numeric_limits::max(); + ASSERT_EQ(ResponsePayloadBytes(integer), 19u); + integer.integer = std::numeric_limits::min(); + ASSERT_EQ(ResponsePayloadBytes(integer), 20u); + } + { + std::string decimal = "-12.5"; + redisReply number = MakeStringReply(REDIS_REPLY_DOUBLE, decimal); + ASSERT_EQ(ResponsePayloadBytes(number), decimal.size()); + } + { + redisReply boolean{}; + boolean.type = REDIS_REPLY_BOOL; + boolean.integer = 1; + ASSERT_EQ(ResponsePayloadBytes(boolean), 1u); + } + { + // PING replies "+PONG\r\n"; `len` is the decoded 4, because the leading '+' + // and the trailing CRLF are framing. + std::string pong = "PONG"; + redisReply status = MakeStringReply(REDIS_REPLY_STATUS, pong); + ASSERT_EQ(ResponsePayloadBytes(status), 4u); + } + { + std::string value(1024, 'x'); + redisReply str = MakeStringReply(REDIS_REPLY_STRING, value); + ASSERT_EQ(ResponsePayloadBytes(str), 1024u); + } + { + // Measured like any other string, but this shape only arises for an error + // nested inside an aggregate reply: RedisResponseFn retries a top-level + // error instead of delivering it, so gcs_redis_response_payload_bytes + // documents error replies as contributing nothing. + std::string error = "ERR unknown command"; + redisReply err = MakeStringReply(REDIS_REPLY_ERROR, error); + ASSERT_EQ(ResponsePayloadBytes(err), error.size()); + } +} + +TEST(TestResponsePayloadBytes, EmptyArrayDoesNotDereferenceElements) { + // hiredis leaves `element` uninitialized when `elements` is 0. Point it at + // garbage to prove the loop bound, not the pointer, is what stops us. + redisReply empty{}; + empty.type = REDIS_REPLY_ARRAY; + empty.elements = 0; + empty.element = reinterpret_cast(0xdeadbeef); + ASSERT_EQ(ResponsePayloadBytes(empty), 0u); +} + +TEST(TestResponsePayloadBytes, HmgetArrayCountsOnlyPresentValues) { + // A partially-missed HMGET: absent fields come back as nil and contribute + // nothing, so the counter measures data actually returned, not requested. + std::string present1(100, 'a'); + std::string present2(250, 'b'); + redisReply value1 = MakeStringReply(REDIS_REPLY_STRING, present1); + redisReply value2 = MakeStringReply(REDIS_REPLY_STRING, present2); + redisReply missing{}; + missing.type = REDIS_REPLY_NIL; + + redisReply *elements[3] = {&value1, &missing, &value2}; + redisReply array = MakeArrayReply(elements, 3); + ASSERT_EQ(ResponsePayloadBytes(array), 350u); +} + +TEST(TestResponsePayloadBytes, ScanArrayCountsCursorAndFieldNames) { + // HSCAN replies [cursor, [field, value, field, value]]. The cursor is a bulk + // string in RESP2 so its bytes count, and the returned field names count too + // -- both facts are stated in the metric description. + std::string cursor = "1234"; + std::string field1 = "field-one"; + std::string value1(64, 'v'); + std::string field2 = "field-two"; + std::string value2(128, 'w'); + + redisReply cursor_reply = MakeStringReply(REDIS_REPLY_STRING, cursor); + redisReply field1_reply = MakeStringReply(REDIS_REPLY_STRING, field1); + redisReply value1_reply = MakeStringReply(REDIS_REPLY_STRING, value1); + redisReply field2_reply = MakeStringReply(REDIS_REPLY_STRING, field2); + redisReply value2_reply = MakeStringReply(REDIS_REPLY_STRING, value2); + + redisReply *pairs[4] = {&field1_reply, &value1_reply, &field2_reply, &value2_reply}; + redisReply pairs_reply = MakeArrayReply(pairs, 4); + redisReply *outer[2] = {&cursor_reply, &pairs_reply}; + redisReply scan_reply = MakeArrayReply(outer, 2); + + const size_t expected = + cursor.size() + field1.size() + value1.size() + field2.size() + value2.size(); + ASSERT_EQ(ResponsePayloadBytes(scan_reply), expected); + ASSERT_EQ(expected, 4u + 9u + 64u + 9u + 128u); +} + +TEST(TestNormalizeRedisCommandLabel, PreservesUppercaseVerb) { + EXPECT_EQ(NormalizeRedisCommandLabel("HGETALL"), "HGETALL"); +} + +TEST(TestNormalizeRedisCommandLabel, UppercasesAsciiVerb) { + EXPECT_EQ(NormalizeRedisCommandLabel("hSeTnX"), "HSETNX"); +} + +TEST(TestNormalizeRedisCommandLabel, TruncatesToMaximumLength) { + const std::string exact(kMaxRedisCommandLabelLength, 'a'); + EXPECT_EQ(NormalizeRedisCommandLabel(exact), + std::string(kMaxRedisCommandLabelLength, 'A')); + + const std::string too_long = exact + "suffix"; + EXPECT_EQ(NormalizeRedisCommandLabel(too_long), + std::string(kMaxRedisCommandLabelLength, 'A')); +} } // namespace ray::gcs diff --git a/src/ray/gcs/store_client/tests/redis_store_client_test.cc b/src/ray/gcs/store_client/tests/redis_store_client_test.cc index 7683d5d5f88c..99fb568f72a4 100644 --- a/src/ray/gcs/store_client/tests/redis_store_client_test.cc +++ b/src/ray/gcs/store_client/tests/redis_store_client_test.cc @@ -19,11 +19,15 @@ #include #include #include +#include #include #include #include #include +#include "absl/cleanup/cleanup.h" +#include "absl/container/flat_hash_set.h" +#include "absl/strings/str_cat.h" #include "gtest/gtest.h" #include "ray/common/test_utils.h" #include "ray/gcs/store_client/tests/store_client_test_base.h" @@ -380,6 +384,370 @@ TEST_F(RedisStoreClientTest, Random) { ASSERT_TRUE(redis_store_client_raw_ptr->pending_redis_request_by_key_.empty()); } +// Tests for the Redis payload byte metrics. These assert exact deltas against +// the definition published with the metrics, so a change here is a change to +// what operators' dashboards mean -- see +// GetGcsRedisRequestPayloadBytesSumMetric in src/ray/gcs/metrics.h. +// +// The fixture drives a real RedisStoreClient with FakeCounters injected, rather +// than inspecting the recording sites, because the contract under test is the +// exported label/value pair and not the code path that produces it. +class RedisStoreClientMetricsTest : public ::testing::Test { + public: + static void SetUpTestCase() { TestSetupUtil::StartUpRedisServers(std::vector()); } + + static void TearDownTestCase() { TestSetupUtil::ShutDownRedisServers(); } + + void SetUp() override { + if (std::getenv("REDIS_CHAOS") != nullptr) { + GTEST_SKIP() << "Exact byte assertions are incompatible with REPLICAOF flapping."; + } + port_ = TEST_REDIS_SERVER_PORTS.front(); + TestSetupUtil::FlushRedisServer(port_); + // The kill switch is read once in the RedisStoreClient constructor, so it + // has to be set before the client is built -- which is also why it is a + // fixture-level knob and not something a test flips mid-run. Rebuilding the + // client while the io thread is polling would destroy a RedisAsyncContext + // out from under a pending socket wait. + RayConfig::instance().gcs_redis_payload_metrics_enabled() = PayloadMetricsEnabled(); + io_service_ = std::make_unique( + /*enable_lag_probe=*/false, /*running_on_single_thread=*/true); + MakeStoreClient(); + thread_ = std::make_unique([this]() { + boost::asio::executor_work_guard work( + io_service_->get_executor()); + io_service_->run(); + }); + } + + void TearDown() override { + if (io_service_ != nullptr) { + io_service_->stop(); + thread_->join(); + store_client_.reset(); + io_service_.reset(); + } + RayConfig::instance().gcs_redis_payload_metrics_enabled() = true; + } + + protected: + // Whether the client under test is built with the payload metrics enabled. + // Overridden by RedisStoreClientMetricsDisabledTest below. + virtual bool PayloadMetricsEnabled() const { return true; } + + void MakeStoreClient() { + RedisClientOptions options{"127.0.0.1", port_}; + store_client_ = std::make_unique( + *io_service_, + options, + clock_, + RedisMetrics{request_bytes_, response_bytes_, command_count_}); + } + + static double ValueFor(const observability::FakeCounter &metric, + const std::string &command, + const std::string &table) { + const absl::flat_hash_map tags{{"Command", command}, + {"TableName", table}}; + auto all = metric.GetTagToValue(); + auto it = all.find(tags); + return it == all.end() ? 0.0 : it->second; + } + + double RequestBytes(const std::string &command, const std::string &table) const { + return ValueFor(request_bytes_, command, table); + } + double ResponseBytes(const std::string &command, const std::string &table) const { + return ValueFor(response_bytes_, command, table); + } + double CommandCount(const std::string &command, const std::string &table) const { + return ValueFor(command_count_, command, table); + } + + // The Redis "key" a GCS table maps to, whose bytes are part of every request. + static std::string RedisKeyFor(const std::string &table) { + return RedisKey{RayConfig::instance().external_storage_namespace(), table}.ToString(); + } + + void PutSync(const std::string &table, const std::string &key, std::string value) { + std::promise promise; + store_client_->AsyncPut( + table, + key, + std::move(value), + /*overwrite=*/true, + {[&promise](bool added) { promise.set_value(added); }, *io_service_}); + promise.get_future().get(); + } + + std::optional GetSync(const std::string &table, const std::string &key) { + std::promise> promise; + store_client_->AsyncGet( + table, + key, + {[&promise](const Status &status, std::optional result) { + RAY_CHECK_OK(status); + promise.set_value(std::move(result)); + }, + *io_service_}); + return promise.get_future().get(); + } + + absl::flat_hash_map MultiGetSync( + const std::string &table, const std::vector &keys) { + std::promise> promise; + store_client_->AsyncMultiGet( + table, + keys, + {[&promise](absl::flat_hash_map result) { + promise.set_value(std::move(result)); + }, + *io_service_}); + return promise.get_future().get(); + } + + absl::flat_hash_map GetAllSync(const std::string &table) { + std::promise> promise; + store_client_->AsyncGetAll( + table, + {[&promise](absl::flat_hash_map result) { + promise.set_value(std::move(result)); + }, + *io_service_}); + return promise.get_future().get(); + } + + int64_t BatchDeleteSync(const std::string &table, + const std::vector &keys) { + std::promise promise; + store_client_->AsyncBatchDelete( + table, keys, {[&promise](int64_t n) { promise.set_value(n); }, *io_service_}); + return promise.get_future().get(); + } + + bool ExistsSync(const std::string &table, const std::string &key) { + std::promise promise; + store_client_->AsyncExists( + table, key, {[&promise](bool e) { promise.set_value(e); }, *io_service_}); + return promise.get_future().get(); + } + + int port_ = 0; + ray::Clock clock_; + std::unique_ptr io_service_; + std::unique_ptr thread_; + std::unique_ptr store_client_; + observability::FakeCounter request_bytes_; + observability::FakeCounter response_bytes_; + observability::FakeCounter command_count_; +}; + +// The published definition, pinned exactly: request bytes are the sum of the +// RESP argument lengths -- verb, Redis key, field name and value -- while the +// HSET integer reply contributes its decimal text length. +TEST_F(RedisStoreClientMetricsTest, PutMatchesDocumentedDefinition) { + const std::string table = "NODE"; + const std::string key = "node-id-0123456789"; + const std::string value(4096, 'v'); + + PutSync(table, key, value); + + const double expected = + std::string("HSET").size() + RedisKeyFor(table).size() + key.size() + value.size(); + EXPECT_EQ(RequestBytes("HSET", table), expected); + EXPECT_EQ(ResponseBytes("HSET", table), 1.0); + EXPECT_EQ(CommandCount("HSET", table), 1.0); +} + +// The issue's size ladder. Each step must move the request counter by exactly +// the value delta, and 8 MiB is far below the 2^53 bound where a double stops +// representing byte counts exactly. +TEST_F(RedisStoreClientMetricsTest, RequestBytesTrackValueSize) { + const std::string table = "ACTOR"; + const std::string key = "k"; + const size_t overhead = + std::string("HSET").size() + RedisKeyFor(table).size() + key.size(); + + double previous = 0; + for (size_t size : {size_t{1} << 10, size_t{1} << 20, size_t{8} << 20}) { + PutSync(table, key, std::string(size, 'x')); + const double now = RequestBytes("HSET", table); + EXPECT_EQ(now - previous, static_cast(overhead + size)) << "size " << size; + previous = now; + } +} + +TEST_F(RedisStoreClientMetricsTest, GetCountsReturnedValueOnly) { + const std::string table = "WORKERS"; + const std::string key = "worker-0"; + const std::string value(2048, 'w'); + PutSync(table, key, value); + + ASSERT_TRUE(GetSync(table, key).has_value()); + EXPECT_EQ(ResponseBytes("HGET", table), static_cast(value.size())); + EXPECT_EQ(RequestBytes("HGET", table), + static_cast(std::string("HGET").size() + RedisKeyFor(table).size() + + key.size())); + EXPECT_EQ(CommandCount("HGET", table), 1.0); + + // A miss still costs a round trip and still sends the key, but returns + // nothing: the counter measures data returned, not data requested. + ASSERT_FALSE(GetSync(table, "absent").has_value()); + EXPECT_EQ(ResponseBytes("HGET", table), static_cast(value.size())); + EXPECT_EQ(CommandCount("HGET", table), 2.0); +} + +// The "bytes sent" half of the issue. An HMGET's field-name vector is the +// payload that grows with a large read, and a values-only definition would +// report zero here. +TEST_F(RedisStoreClientMetricsTest, MultiGetCountsRequestFieldNames) { + const std::string table = "KV"; + std::vector keys; + size_t key_bytes = 0; + for (int i = 0; i < 50; ++i) { + keys.push_back(absl::StrCat("a-fairly-long-internal-kv-field-name-", i)); + key_bytes += keys.back().size(); + } + + // Nothing was written, so every field misses and the reply is all nils. + ASSERT_TRUE(MultiGetSync(table, keys).empty()); + + EXPECT_EQ(RequestBytes("HMGET", table), + static_cast(std::string("HMGET").size() + RedisKeyFor(table).size() + + key_bytes)); + EXPECT_EQ(ResponseBytes("HMGET", table), 0.0); + EXPECT_EQ(CommandCount("HMGET", table), 1.0); +} + +// The issue's explicit expectation: HSCAN must report non-zero response bytes. +// This is also the regression test for reading a length off a string that has +// already been moved into the result map -- that mistake reports ~0 here. +TEST_F(RedisStoreClientMetricsTest, GetAllReportsScanResponseBytes) { + const std::string table = "PLACEMENT_GROUP"; + const size_t num_entries = 200; + size_t field_and_value_bytes = 0; + for (size_t i = 0; i < num_entries; ++i) { + std::string key = absl::StrCat("placement-group-", i); + std::string value(512, 'p'); + field_and_value_bytes += key.size() + value.size(); + PutSync(table, key, value); + } + + ASSERT_EQ(GetAllSync(table).size(), num_entries); + + // Strictly greater than the fields and values, because every HSCAN round also + // returns a cursor bulk string. Any zero, or any values-only accounting, + // fails here. + EXPECT_GT(ResponseBytes("HSCAN", table), static_cast(field_and_value_bytes)); + EXPECT_GE(CommandCount("HSCAN", table), 1.0); + EXPECT_GT(RequestBytes("HSCAN", table), 0.0); +} + +// Batched operations are counted per chunk, so the count is Redis round trips +// rather than StoreClient calls. Pin that, and pin that the byte totals +// aggregate across chunks. +TEST_F(RedisStoreClientMetricsTest, BatchedCommandsCountPerChunk) { + const std::string table = "JOB"; + const size_t batch = 4; + RayConfig::instance().maximum_gcs_storage_operation_batch_size() = batch; + auto restore = absl::MakeCleanup( + []() { RayConfig::instance().maximum_gcs_storage_operation_batch_size() = 1000; }); + + std::vector keys; + size_t key_bytes = 0; + for (size_t i = 0; i < 10; ++i) { + keys.push_back(absl::StrCat("job-", i)); + key_bytes += keys.back().size(); + } + + MultiGetSync(table, keys); + const double expected_chunks = 3; // ceil(10 / 4) + EXPECT_EQ(CommandCount("HMGET", table), expected_chunks); + EXPECT_EQ(RequestBytes("HMGET", table), + static_cast(expected_chunks * (std::string("HMGET").size() + + RedisKeyFor(table).size()) + + key_bytes)); + + BatchDeleteSync(table, keys); + EXPECT_EQ(CommandCount("HDEL", table), expected_chunks); +} + +// Every public method should use the actual Redis verb and the logical table, +// never a rendered Redis key containing the per-cluster namespace. +TEST_F(RedisStoreClientMetricsTest, LabelsMatchCommandsAndTables) { + const std::string table = "ACTOR_TASK_SPEC"; + PutSync(table, "k1", "v1"); + GetSync(table, "k1"); + MultiGetSync(table, {"k1", "k2"}); + GetAllSync(table); + ExistsSync(table, "k1"); + BatchDeleteSync(table, {"k1"}); + { + std::promise promise; + store_client_->AsyncGetNextJobID( + {[&promise](int id) { promise.set_value(id); }, *io_service_}); + promise.get_future().get(); + } + { + std::promise promise; + store_client_->AsyncCheckHealth( + {[&promise](Status s) { promise.set_value(s); }, *io_service_}); + ASSERT_TRUE(promise.get_future().get().ok()); + } + + const absl::flat_hash_set allowed_commands{"HSET", + "HSETNX", + "HGET", + "HMGET", + "HDEL", + "HEXISTS", + "HSCAN", + "INCRBY", + "PING", + "SCAN", + "DEL", + "UNLINK", + "INFO"}; + const absl::flat_hash_set allowed_tables{ + table, "JobCounter", std::string(kNoTable), std::string(kAllTables)}; + + size_t observed = 0; + for (const auto *metric : {&request_bytes_, &response_bytes_, &command_count_}) { + for (const auto &[tags, _] : metric->GetTagToValue()) { + ASSERT_EQ(tags.size(), 2u); + const std::string &command = tags.at("Command"); + const std::string &table_name = tags.at("TableName"); + EXPECT_TRUE(allowed_commands.contains(command)) << "unexpected Command " << command; + EXPECT_TRUE(allowed_tables.contains(table_name)) + << "unexpected TableName " << table_name; + // The rendered Redis key carries the storage namespace, which is a + // per-cluster identifier. If it ever leaks into a label the cardinality + // bound is gone. + EXPECT_EQ(table_name.find(RayConfig::instance().external_storage_namespace()), + std::string::npos); + ++observed; + } + } + EXPECT_GT(observed, 0u); +} + +// Same fixture, but the client is constructed with the kill switch already off, +// so nothing has to be torn down mid-test to observe the disabled behavior. +class RedisStoreClientMetricsDisabledTest : public RedisStoreClientMetricsTest { + protected: + bool PayloadMetricsEnabled() const override { return false; } +}; + +TEST_F(RedisStoreClientMetricsDisabledTest, KillSwitchStopsRecording) { + const std::string table = "NODE"; + PutSync(table, "k", std::string(1024, 'v')); + ASSERT_TRUE(GetSync(table, "k").has_value()); + + EXPECT_TRUE(request_bytes_.GetTagToValue().empty()); + EXPECT_TRUE(response_bytes_.GetTagToValue().empty()); + EXPECT_TRUE(command_count_.GetTagToValue().empty()); +} + // Tests for RedisDelKeyPrefixSync (namespace cleanup). These assert exact // command-count deltas from INFO commandstats: RedisDelKeyPrefixSync runs its // own Connect(), and every non-Sentinel connect issues one DEL DUMMY, so @@ -436,10 +804,12 @@ class RedisDelKeyPrefixSyncTest : public ::testing::Test { // Runs one command on the admin connection and returns its reply. std::shared_ptr RunCmd(std::vector cmd) { std::promise> promise; - context_->RunArgvAsync(std::move(cmd), - [&promise](const std::shared_ptr &reply) { - promise.set_value(reply); - }); + context_->RunArgvAsync( + std::move(cmd), + [&promise](const std::shared_ptr &reply) { + promise.set_value(reply); + }, + kAllTables); return promise.get_future().get(); } diff --git a/src/ray/gcs/tests/gcs_server_rpc_test.cc b/src/ray/gcs/tests/gcs_server_rpc_test.cc index 45220752b465..e92a35b64575 100644 --- a/src/ray/gcs/tests/gcs_server_rpc_test.cc +++ b/src/ray/gcs/tests/gcs_server_rpc_test.cc @@ -55,6 +55,9 @@ class GcsServerTest : public ::testing::Test { /*storage_operation_latency_in_ms_histogram=*/ storage_operation_latency_in_ms_histogram_, /*storage_operation_count_counter=*/storage_operation_count_counter_, + /*redis_request_payload_bytes_sum=*/redis_request_payload_bytes_sum_, + /*redis_response_payload_bytes_sum=*/redis_response_payload_bytes_sum_, + /*redis_command_count_counter=*/redis_command_count_counter_, /*resource_usage_gauge=*/fake_resource_usage_gauge_, fake_scheduler_placement_time_ms_histogram_, /*health_check_rpc_latency_ms_histogram=*/ @@ -323,6 +326,9 @@ class GcsServerTest : public ::testing::Test { observability::FakeGauge task_events_stored_gauge_; observability::FakeHistogram storage_operation_latency_in_ms_histogram_; observability::FakeCounter storage_operation_count_counter_; + observability::FakeCounter redis_request_payload_bytes_sum_; + observability::FakeCounter redis_response_payload_bytes_sum_; + observability::FakeCounter redis_command_count_counter_; observability::FakeCounter fake_dropped_events_counter_; observability::FakeGauge fake_resource_usage_gauge_; observability::FakeHistogram fake_scheduler_placement_time_ms_histogram_; diff --git a/src/ray/gcs_rpc_client/tests/gcs_client_reconnection_test.cc b/src/ray/gcs_rpc_client/tests/gcs_client_reconnection_test.cc index 18f589d82910..28b048d6512a 100644 --- a/src/ray/gcs_rpc_client/tests/gcs_client_reconnection_test.cc +++ b/src/ray/gcs_rpc_client/tests/gcs_client_reconnection_test.cc @@ -65,6 +65,9 @@ class GcsClientReconnectionTest : public ::testing::Test { /*storage_operation_latency_in_ms_histogram=*/ storage_operation_latency_in_ms_histogram_, /*storage_operation_count_counter=*/storage_operation_count_counter_, + /*redis_request_payload_bytes_sum=*/redis_request_payload_bytes_sum_, + /*redis_response_payload_bytes_sum=*/redis_response_payload_bytes_sum_, + /*redis_command_count_counter=*/redis_command_count_counter_, /*resource_usage_gauge=*/fake_resource_usage_gauge_, scheduler_placement_time_ms_histogram_, /*health_check_rpc_latency_ms_histogram=*/ @@ -205,6 +208,9 @@ class GcsClientReconnectionTest : public ::testing::Test { observability::FakeGauge task_events_stored_gauge_; observability::FakeHistogram storage_operation_latency_in_ms_histogram_; observability::FakeCounter storage_operation_count_counter_; + observability::FakeCounter redis_request_payload_bytes_sum_; + observability::FakeCounter redis_response_payload_bytes_sum_; + observability::FakeCounter redis_command_count_counter_; observability::FakeCounter fake_dropped_events_counter_; observability::FakeGauge fake_resource_usage_gauge_; observability::FakeHistogram scheduler_placement_time_ms_histogram_; diff --git a/src/ray/gcs_rpc_client/tests/gcs_client_test.cc b/src/ray/gcs_rpc_client/tests/gcs_client_test.cc index 4d404a9677fd..5042e0839630 100644 --- a/src/ray/gcs_rpc_client/tests/gcs_client_test.cc +++ b/src/ray/gcs_rpc_client/tests/gcs_client_test.cc @@ -105,6 +105,9 @@ class GcsClientTest : public ::testing::TestWithParam { /*storage_operation_latency_in_ms_histogram=*/ storage_operation_latency_in_ms_histogram_, /*storage_operation_count_counter=*/storage_operation_count_counter_, + /*redis_request_payload_bytes_sum=*/redis_request_payload_bytes_sum_, + /*redis_response_payload_bytes_sum=*/redis_response_payload_bytes_sum_, + /*redis_command_count_counter=*/redis_command_count_counter_, /*resource_usage_gauge=*/fake_resource_usage_gauge_, scheduler_placement_time_ms_histogram_, /*health_check_rpc_latency_ms_histogram=*/ @@ -200,6 +203,9 @@ class GcsClientTest : public ::testing::TestWithParam { /*storage_operation_latency_in_ms_histogram=*/ storage_operation_latency_in_ms_histogram_, /*storage_operation_count_counter=*/storage_operation_count_counter_, + /*redis_request_payload_bytes_sum=*/redis_request_payload_bytes_sum_, + /*redis_response_payload_bytes_sum=*/redis_response_payload_bytes_sum_, + /*redis_command_count_counter=*/redis_command_count_counter_, /*resource_usage_gauge=*/fake_resource_usage_gauge_, scheduler_placement_time_ms_histogram_, /*health_check_rpc_latency_ms_histogram=*/ @@ -511,6 +517,9 @@ class GcsClientTest : public ::testing::TestWithParam { observability::FakeGauge task_events_stored_gauge_; observability::FakeHistogram storage_operation_latency_in_ms_histogram_; observability::FakeCounter storage_operation_count_counter_; + observability::FakeCounter redis_request_payload_bytes_sum_; + observability::FakeCounter redis_response_payload_bytes_sum_; + observability::FakeCounter redis_command_count_counter_; observability::FakeCounter fake_dropped_events_counter_; observability::FakeGauge fake_resource_usage_gauge_; observability::FakeHistogram scheduler_placement_time_ms_histogram_; diff --git a/src/ray/gcs_rpc_client/tests/global_state_accessor_test.cc b/src/ray/gcs_rpc_client/tests/global_state_accessor_test.cc index 506aff8a6a52..268f8dab930d 100644 --- a/src/ray/gcs_rpc_client/tests/global_state_accessor_test.cc +++ b/src/ray/gcs_rpc_client/tests/global_state_accessor_test.cc @@ -86,6 +86,9 @@ class GlobalStateAccessorTest : public ::testing::TestWithParam { /*storage_operation_latency_in_ms_histogram=*/ fake_storage_operation_latency_in_ms_histogram_, /*storage_operation_count_counter=*/fake_storage_operation_count_counter_, + /*redis_request_payload_bytes_sum=*/fake_redis_request_payload_bytes_sum_, + /*redis_response_payload_bytes_sum=*/fake_redis_response_payload_bytes_sum_, + /*redis_command_count_counter=*/fake_redis_command_count_counter_, /*resource_usage_gauge=*/fake_resource_usage_gauge_, fake_scheduler_placement_time_ms_histogram_, /*health_check_rpc_latency_ms_histogram=*/ @@ -166,6 +169,9 @@ class GlobalStateAccessorTest : public ::testing::TestWithParam { observability::FakeGauge fake_task_events_stored_gauge_; observability::FakeHistogram fake_storage_operation_latency_in_ms_histogram_; observability::FakeCounter fake_storage_operation_count_counter_; + observability::FakeCounter fake_redis_request_payload_bytes_sum_; + observability::FakeCounter fake_redis_response_payload_bytes_sum_; + observability::FakeCounter fake_redis_command_count_counter_; observability::FakeGauge fake_resource_usage_gauge_; observability::FakeHistogram fake_scheduler_placement_time_ms_histogram_; observability::FakeHistogram fake_health_check_rpc_latency_ms_histogram_;