[core][gcs] Export Redis payload byte metrics by command and table #65554
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -15,9 +15,13 @@ | |
| #include "ray/gcs/store_client/redis_context.h" | ||
|
|
||
| #include <cerrno> | ||
| #include <charconv> | ||
| #include <cstddef> | ||
| #include <limits> | ||
| #include <memory> | ||
| #include <sstream> | ||
| #include <string> | ||
| #include <string_view> | ||
| #include <utility> | ||
| #include <vector> | ||
|
|
||
|
|
@@ -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<size_t>(reply.len); | ||
| case REDIS_REPLY_INTEGER: { | ||
| char buffer[std::numeric_limits<long long>::digits10 + 3]; | ||
| const auto [end, error] = | ||
| std::to_chars(buffer, buffer + sizeof(buffer), reply.integer); | ||
| RAY_CHECK(error == std::errc{}); | ||
| return static_cast<size_t>(end - buffer); | ||
| } | ||
| case REDIS_REPLY_DOUBLE: | ||
| // hiredis preserves the original RESP3 decimal representation in `str`. | ||
| return static_cast<size_t>(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]); | ||
| } | ||
|
Comment on lines
+79
to
+82
Contributor
There was a problem hiding this comment. In defensive programming, it is important to handle potential null pointers when dealing with external library structures like size_t total = 0;
if (reply.elements > 0 && reply.element != nullptr) {
for (size_t i = 0; i < reply.elements; ++i) {
if (reply.element[i] != nullptr) {
total += ResponsePayloadBytes(*reply.element[i]);
}
}
}
Contributor
Author
There was a problem hiding this comment. Thanks for looking at this. I dug into it and I'm going to leave it as-is, for two reasons. The guard wouldn't actually prevent the crash.
The one case that genuinely is unspecified — hiredis leaves Happy to add the checks if a maintainer prefers defense-in-depth here, but I'd want to add them to |
||
| return total; | ||
| } | ||
| default: | ||
| // Unknown reply types have no payload definition. | ||
| return 0; | ||
|
Contributor
There was a problem hiding this comment. Is it correct to use 0 for these types? For integers, shouldn't we count the decimal length?
Contributor
Author
There was a problem hiding this comment. You're right. The decimal representation between the RESP integer prefix and CRLF is application payload, so returning zero undercounted it. Integer replies now count their decimal text length, including a minus sign, using allocation-free I also audited the other scalar reply types: doubles use the original text length preserved by hiredis, booleans contribute one byte, and nil remains zero. Tests cover zero, negative values, Updated in 6400b8f. |
||
| } | ||
| } | ||
|
|
||
| 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<std::string> args, | ||
| 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), | ||
| 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<double>(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<double>(ResponsePayloadBytes(*redis_reply)), | ||
| {{"Command", request_cxt->command_label_}, | ||
| {"TableName", request_cxt->table_label_}}); | ||
| } | ||
| auto reply = std::make_shared<CallbackReply>(*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, | ||
| std::optional<RedisMetrics> 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<CallbackReply> RedisContext::RunArgvSync( | |
| } | ||
|
|
||
| void RedisContext::RunArgvAsync(std::vector<std::string> args, | ||
| RedisCallback redis_callback, | ||
| std::string_view table_label) { | ||
| RAY_CHECK(redis_async_context_); | ||
| 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(); | ||
| } | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.