From 0a27d1a0e8f117199df57cc6962adae71ff48588 Mon Sep 17 00:00:00 2001 From: gavinchou Date: Tue, 2 Jun 2026 20:51:15 +0800 Subject: [PATCH 1/4] Add S3 rate limit observability --- be/src/common/config.cpp | 3 ++ be/src/common/config.h | 1 + be/src/io/fs/azure_obj_storage_client.cpp | 4 +- be/src/io/fs/s3_obj_storage_client.cpp | 20 +++++++- be/src/util/s3_util.cpp | 61 ++++++++++++++++++++--- be/src/util/s3_util.h | 1 + cloud/test/s3_rate_limiter_test.cpp | 25 +++++++--- common/cpp/obj_retry_strategy.cpp | 19 ++++++- common/cpp/obj_retry_strategy.h | 4 +- common/cpp/token_bucket_rate_limiter.h | 14 ++++-- 10 files changed, 128 insertions(+), 24 deletions(-) diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index 7b6db1636234aa..2564e45b8401a5 100644 --- a/be/src/common/config.cpp +++ b/be/src/common/config.cpp @@ -1513,6 +1513,9 @@ DEFINE_mInt64(s3_put_token_per_second, "1000000000000000000"); DEFINE_Validator(s3_put_token_per_second, [](int64_t config) -> bool { return config > 0; }); DEFINE_mInt64(s3_put_token_limit, "0"); +// Log active S3 rate limiter every N throttled/rejected requests, 0 means no log. +DEFINE_mInt64(s3_rate_limiter_log_interval, "1000"); +DEFINE_Validator(s3_rate_limiter_log_interval, [](int64_t config) -> bool { return config >= 0; }); DEFINE_String(trino_connector_plugin_dir, "${DORIS_HOME}/plugins/connectors"); diff --git a/be/src/common/config.h b/be/src/common/config.h index 427282a4452bc4..aa9f9aa61b1076 100644 --- a/be/src/common/config.h +++ b/be/src/common/config.h @@ -1598,6 +1598,7 @@ DECLARE_mInt64(s3_get_token_limit); DECLARE_mInt64(s3_put_bucket_tokens); DECLARE_mInt64(s3_put_token_per_second); DECLARE_mInt64(s3_put_token_limit); +DECLARE_mInt64(s3_rate_limiter_log_interval); // max s3 client retry times DECLARE_mInt32(max_s3_client_retry); // When meet s3 429 error, the "get" request will diff --git a/be/src/io/fs/azure_obj_storage_client.cpp b/be/src/io/fs/azure_obj_storage_client.cpp index 85b81c1deb23df..8b09a2a4aafd11 100644 --- a/be/src/io/fs/azure_obj_storage_client.cpp +++ b/be/src/io/fs/azure_obj_storage_client.cpp @@ -43,6 +43,7 @@ #include "common/exception.h" #include "common/logging.h" #include "common/status.h" +#include "cpp/obj_retry_strategy.h" #include "io/fs/obj_storage_client.h" #include "util/bvar_helper.h" #include "util/coding.h" @@ -74,7 +75,7 @@ auto s3_rate_limit(doris::S3RateLimitType op, Func callback) -> decltype(callbac if (!doris::config::enable_s3_rate_limiter) { return callback(); } - auto sleep_duration = doris::S3ClientFactory::instance().rate_limiter(op)->add(1); + auto sleep_duration = doris::apply_s3_rate_limit(op); if (sleep_duration < 0) { throw std::runtime_error("Azure exceeds request limit"); } @@ -124,6 +125,7 @@ ObjectStorageResponse do_azure_client_call(Func f, const ObjectStoragePathOption try { f(); } catch (Azure::Core::RequestFailedException& e) { + doris::record_object_request_failed(static_cast(e.StatusCode)); auto tls_debug_suffix = build_azure_tls_debug_suffix( fmt::format("{} {}", e.what(), e.Message), tls_debug_context); auto msg = fmt::format( diff --git a/be/src/io/fs/s3_obj_storage_client.cpp b/be/src/io/fs/s3_obj_storage_client.cpp index 1b8bf7b473c076..f9ed8e155ff59c 100644 --- a/be/src/io/fs/s3_obj_storage_client.cpp +++ b/be/src/io/fs/s3_obj_storage_client.cpp @@ -66,6 +66,7 @@ #include "common/logging.h" #include "common/status.h" +#include "cpp/obj_retry_strategy.h" #include "cpp/sync_point.h" #include "io/fs/err_utils.h" #include "io/fs/s3_common.h" @@ -82,7 +83,7 @@ auto s3_rate_limit(doris::S3RateLimitType op, Func callback) -> decltype(callbac if (!doris::config::enable_s3_rate_limiter) { return callback(); } - auto sleep_duration = doris::S3ClientFactory::instance().rate_limiter(op)->add(1); + auto sleep_duration = doris::apply_s3_rate_limit(op); if (sleep_duration < 0) { return T(s3_error_factory()); } @@ -98,6 +99,10 @@ template auto s3_put_rate_limit(Func callback) -> decltype(callback()) { return s3_rate_limit(doris::S3RateLimitType::PUT, std::move(callback)); } + +void record_s3_request_failed(const Aws::S3::S3Error& error) { + doris::record_object_request_failed(static_cast(error.GetResponseCode())); +} } // namespace namespace Aws::S3::Model { @@ -140,6 +145,7 @@ ObjectStorageUploadResponse S3ObjStorageClient::create_multipart_upload( << ", request_id=" << request_id << ", bucket=" << opts.bucket << ", key=" << opts.key; if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); auto st = s3fs_error(outcome.GetError(), fmt::format("failed to CreateMultipartUpload: {} ", opts.path.native())); LOG(WARNING) << st << " request_id=" << request_id; @@ -177,6 +183,7 @@ ObjectStorageResponse S3ObjStorageClient::put_object(const ObjectStoragePathOpti : outcome.GetError().GetRequestId(); if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); auto st = s3fs_error(outcome.GetError(), fmt::format("failed to put object: {}", opts.path.native())); LOG(WARNING) << st << ", request_id=" << request_id; @@ -222,6 +229,7 @@ ObjectStorageUploadResponse S3ObjStorageClient::upload_part(const ObjectStorageP TEST_SYNC_POINT_CALLBACK("S3FileWriter::_upload_one_part", &outcome); if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); auto st = Status::IOError( "failed to UploadPart bucket={}, key={}, part_num={}, upload_id={}, message={}, " "exception_name={}, response_code={}, request_id={}", @@ -275,6 +283,7 @@ ObjectStorageResponse S3ObjStorageClient::complete_multipart_upload( : outcome.GetError().GetRequestId(); if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); auto st = s3fs_error(outcome.GetError(), fmt::format("failed to CompleteMultipartUpload: {}, upload_id={}", opts.path.native(), *opts.upload_id)); @@ -305,6 +314,7 @@ ObjectStorageHeadResponse S3ObjStorageClient::head_object(const ObjectStoragePat } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { return {.resp = {convert_to_obj_response(Status::Error(""))}}; } else { + record_s3_request_failed(outcome.GetError()); return {.resp = {convert_to_obj_response( s3fs_error(outcome.GetError(), fmt::format("failed to check exists {}", opts.key))), @@ -324,6 +334,7 @@ ObjectStorageResponse S3ObjStorageClient::get_object(const ObjectStoragePathOpti SCOPED_BVAR_LATENCY(s3_bvar::s3_get_latency); auto outcome = s3_get_rate_limit([&]() { return _client->GetObject(request); }); if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); return {convert_to_obj_response(s3fs_error( outcome.GetError(), fmt::format("failed to read from {}", opts.key))), static_cast(outcome.GetError().GetResponseCode()), @@ -362,6 +373,7 @@ ObjectStorageResponse S3ObjStorageClient::list_objects(const ObjectStoragePathOp return ObjectStorageResponse::OK(); } + record_s3_request_failed(outcome.GetError()); return {convert_to_obj_response(s3fs_error( outcome.GetError(), fmt::format("failed to list {}", opts.prefix))), static_cast(outcome.GetError().GetResponseCode()), @@ -406,6 +418,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects(const ObjectStoragePath auto delete_outcome = s3_put_rate_limit([&]() { return _client->DeleteObjects(delete_request); }); if (!delete_outcome.IsSuccess()) { + record_s3_request_failed(delete_outcome.GetError()); return {convert_to_obj_response( s3fs_error(delete_outcome.GetError(), fmt::format("failed to delete dir {}", opts.key))), @@ -433,6 +446,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_object(const ObjectStoragePathO outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { return ObjectStorageResponse::OK(); } + record_s3_request_failed(outcome.GetError()); return {convert_to_obj_response(s3fs_error(outcome.GetError(), fmt::format("failed to delete file {}", opts.key))), static_cast(outcome.GetError().GetResponseCode()), @@ -453,6 +467,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects_recursively( outcome = s3_get_rate_limit([&]() { return _client->ListObjectsV2(request); }); } if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); return {convert_to_obj_response(s3fs_error( outcome.GetError(), fmt::format("failed to list objects when delete dir {}", opts.prefix))), @@ -473,6 +488,7 @@ ObjectStorageResponse S3ObjStorageClient::delete_objects_recursively( auto delete_outcome = s3_put_rate_limit([&]() { return _client->DeleteObjects(delete_request); }); if (!delete_outcome.IsSuccess()) { + record_s3_request_failed(delete_outcome.GetError()); return {convert_to_obj_response( s3fs_error(delete_outcome.GetError(), fmt::format("failed to delete dir {}", opts.key))), @@ -502,4 +518,4 @@ std::string S3ObjStorageClient::generate_presigned_url(const ObjectStoragePathOp expiration_secs); } -} // namespace doris::io \ No newline at end of file +} // namespace doris::io diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index 851fbf43edf5e3..b444b094a7c62b 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -158,10 +158,17 @@ constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID"; constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE"; } // namespace -bvar::Adder get_rate_limit_ns("get_rate_limit_ns"); -bvar::Adder get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num"); -bvar::Adder put_rate_limit_ns("put_rate_limit_ns"); -bvar::Adder put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num"); +bvar::Adder s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns"); +bvar::Adder s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count"); +bvar::Adder s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count"); +bvar::Adder s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns"); +bvar::Adder s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count"); +bvar::Adder s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count"); + +static std::atomic s3_get_rate_limit_sleep_log_count {0}; +static std::atomic s3_get_rate_limit_rejected_log_count {0}; +static std::atomic s3_put_rate_limit_sleep_log_count {0}; +static std::atomic s3_put_rate_limit_rejected_log_count {0}; static std::atomic last_s3_get_token_bucket_tokens {0}; static std::atomic last_s3_get_token_limit {0}; @@ -230,6 +237,44 @@ int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_bur return S3ClientFactory::instance().rate_limiter(type)->reset(max_speed, max_burst, limit); } +int64_t apply_s3_rate_limit(S3RateLimitType type) { + auto sleep_duration = S3ClientFactory::instance().rate_limiter(type)->add(1); + int64_t interval = config::s3_rate_limiter_log_interval; + if (interval <= 0 || sleep_duration == 0) { + return sleep_duration; + } + + auto is_get = type == S3RateLimitType::GET; + auto* sleep_log_count = + is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count; + auto* rejected_log_count = + is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count; + int64_t token_per_second = + is_get ? config::s3_get_token_per_second : config::s3_put_token_per_second; + int64_t bucket_tokens = is_get ? config::s3_get_bucket_tokens : config::s3_put_bucket_tokens; + int64_t token_limit = is_get ? config::s3_get_token_limit : config::s3_put_token_limit; + + if (sleep_duration > 0) { + int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1; + if (count == 1 || count % interval == 0) { + LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter" + << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count + << ", token_per_second=" << token_per_second + << ", bucket_tokens=" << bucket_tokens << ", token_limit=" << token_limit; + } + } else { + int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1; + if (count == 1 || count % interval == 0) { + LOG(WARNING) << "S3 " << to_string(type) + << " request is rejected by local rate limiter" + << ", rejected_count=" << count + << ", token_per_second=" << token_per_second + << ", bucket_tokens=" << bucket_tokens << ", token_limit=" << token_limit; + } + } + return sleep_duration; +} + S3ClientFactory::S3ClientFactory() { _aws_options = Aws::SDKOptions {}; auto logLevel = static_cast(config::aws_log_level); @@ -243,11 +288,15 @@ S3ClientFactory::S3ClientFactory() { std::make_unique( config::s3_get_token_per_second, config::s3_get_bucket_tokens, config::s3_get_token_limit, - metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)), + metric_func_factory(s3_get_rate_limit_sleep_ns, + s3_get_rate_limit_sleep_count, + &s3_get_rate_limit_rejected_count)), std::make_unique( config::s3_put_token_per_second, config::s3_put_bucket_tokens, config::s3_put_token_limit, - metric_func_factory(put_rate_limit_ns, put_rate_limit_exceed_req_num))}; + metric_func_factory(s3_put_rate_limit_sleep_ns, + s3_put_rate_limit_sleep_count, + &s3_put_rate_limit_rejected_count))}; #ifdef USE_AZURE auto azureLogLevel = diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h index 636c78f1b65520..62f50e31a2217b 100644 --- a/be/src/util/s3_util.h +++ b/be/src/util/s3_util.h @@ -64,6 +64,7 @@ extern bvar::LatencyRecorder s3_copy_object_latency; std::string hide_access_key(const std::string& ak); int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit); +int64_t apply_s3_rate_limit(S3RateLimitType type); class S3URI; struct S3ClientConf { diff --git a/cloud/test/s3_rate_limiter_test.cpp b/cloud/test/s3_rate_limiter_test.cpp index b1fd5a8115fa6b..20496752e586cb 100644 --- a/cloud/test/s3_rate_limiter_test.cpp +++ b/cloud/test/s3_rate_limiter_test.cpp @@ -107,18 +107,27 @@ TEST(S3RateLimiterTest, ExceedLimit) { } TEST(S3RateLimiterHolderTest, BvarMetric) { - bvar::Adder rate_limit_ns("rate_limit_ns"); - bvar::Adder rate_limit_exceed_req_num("rate_limit_exceed_req_num"); + bvar::Adder rate_limit_sleep_ns("rate_limit_sleep_ns"); + bvar::Adder rate_limit_sleep_count("rate_limit_sleep_count"); + bvar::Adder rate_limit_rejected_count("rate_limit_rejected_count"); auto rate_limiter_holder = doris::S3RateLimiterHolder( - 125, 250, 500, doris::metric_func_factory(rate_limit_ns, rate_limit_exceed_req_num)); + 125, 250, 251, + doris::metric_func_factory(rate_limit_sleep_ns, rate_limit_sleep_count, + &rate_limit_rejected_count)); int64_t sleep_time = rate_limiter_holder.add(250); EXPECT_EQ(sleep_time, 0); - EXPECT_EQ(rate_limit_ns.get_value(), 0); - EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 0); + EXPECT_EQ(rate_limit_sleep_ns.get_value(), 0); + EXPECT_EQ(rate_limit_sleep_count.get_value(), 0); + EXPECT_EQ(rate_limit_rejected_count.get_value(), 0); sleep_time = rate_limiter_holder.add(1); - EXPECT_GT(rate_limit_ns.get_value(), 0); - EXPECT_EQ(rate_limit_exceed_req_num.get_value(), 1); EXPECT_GT(sleep_time, 0); -} \ No newline at end of file + EXPECT_GT(rate_limit_sleep_ns.get_value(), 0); + EXPECT_EQ(rate_limit_sleep_count.get_value(), 1); + EXPECT_EQ(rate_limit_rejected_count.get_value(), 0); + + sleep_time = rate_limiter_holder.add(1); + EXPECT_EQ(sleep_time, -1); + EXPECT_EQ(rate_limit_rejected_count.get_value(), 1); +} diff --git a/common/cpp/obj_retry_strategy.cpp b/common/cpp/obj_retry_strategy.cpp index 6da4c23980a9b7..154a8f010ca017 100644 --- a/common/cpp/obj_retry_strategy.cpp +++ b/common/cpp/obj_retry_strategy.cpp @@ -24,6 +24,16 @@ namespace doris { bvar::Adder object_request_retry_count("object_request_retry_count"); +bvar::Adder s3_request_retry_too_many_requests_count( + "s3_request_retry_too_many_requests_count"); +bvar::Adder s3_request_failed_too_many_requests_count( + "s3_request_failed_too_many_requests_count"); + +void record_object_request_failed(int http_code) { + if (http_code == static_cast(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) { + s3_request_failed_too_many_requests_count << 1; + } +} S3CustomRetryStrategy::S3CustomRetryStrategy(int maxRetries) : DefaultRetryStrategy(maxRetries) {} @@ -37,6 +47,9 @@ bool S3CustomRetryStrategy::ShouldRetry(const Aws::Client::AWSError AzureRetryRecordPolicy::Send( static_cast(response->GetStatusCode()) < 200) { if (retry_count > 0) { object_request_retry_count << 1; + if (static_cast(response->GetStatusCode()) == + static_cast(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) { + s3_request_retry_too_many_requests_count << 1; + } } // If the response is not successful, we log the retry attempt and status code. @@ -74,4 +91,4 @@ std::unique_ptr AzureRetryRecordPolicy::Clon return std::make_unique(*this); } #endif -} // namespace doris \ No newline at end of file +} // namespace doris diff --git a/common/cpp/obj_retry_strategy.h b/common/cpp/obj_retry_strategy.h index dd98f871716fe7..5e7dbcaad79795 100644 --- a/common/cpp/obj_retry_strategy.h +++ b/common/cpp/obj_retry_strategy.h @@ -25,6 +25,8 @@ #endif namespace doris { +void record_object_request_failed(int http_code); + class S3CustomRetryStrategy final : public Aws::Client::DefaultRetryStrategy { public: S3CustomRetryStrategy(int maxRetries); @@ -47,4 +49,4 @@ class AzureRetryRecordPolicy final : public Azure::Core::Http::Policies::HttpPol Azure::Core::Context const& context) const override; }; #endif -} // namespace doris \ No newline at end of file +} // namespace doris diff --git a/common/cpp/token_bucket_rate_limiter.h b/common/cpp/token_bucket_rate_limiter.h index dbeca3b897db03..a2f571dc1fa9e0 100644 --- a/common/cpp/token_bucket_rate_limiter.h +++ b/common/cpp/token_bucket_rate_limiter.h @@ -32,11 +32,15 @@ enum class S3RateLimitType : int { extern std::string to_string(S3RateLimitType type); extern S3RateLimitType string_to_s3_rate_limit_type(std::string_view value); -inline auto metric_func_factory(bvar::Adder& ns_bvar, bvar::Adder& req_num_bvar) { - return [&](int64_t ns) { +inline auto metric_func_factory(bvar::Adder& sleep_ns_bvar, + bvar::Adder& sleep_count_bvar, + bvar::Adder* rejected_count_bvar = nullptr) { + return [&, rejected_count_bvar](int64_t ns) { if (ns > 0) { - ns_bvar << ns; - req_num_bvar << 1; + sleep_ns_bvar << ns; + sleep_count_bvar << 1; + } else if (ns < 0 && rejected_count_bvar != nullptr) { + *rejected_count_bvar << 1; } }; } @@ -96,4 +100,4 @@ class TokenBucketRateLimiterHolder { using S3RateLimiter = TokenBucketRateLimiter; using S3RateLimiterHolder = TokenBucketRateLimiterHolder; -} // namespace doris \ No newline at end of file +} // namespace doris From ac67923b29f5b5d590584269cac04c105bb87e10 Mon Sep 17 00:00:00 2001 From: gavinchou Date: Wed, 3 Jun 2026 02:45:35 +0800 Subject: [PATCH 2/4] Address S3 rate limit observability review comments --- be/src/util/s3_util.cpp | 58 ++-------------------- cloud/src/common/config.h | 3 ++ cloud/src/recycler/azure_obj_client.cpp | 9 +++- cloud/src/recycler/s3_accessor.cpp | 10 +--- cloud/src/recycler/s3_obj_client.cpp | 29 ++++++++--- cloud/test/s3_rate_limiter_test.cpp | 20 ++++++++ common/cpp/token_bucket_rate_limiter.cpp | 62 ++++++++++++++++++++++++ common/cpp/token_bucket_rate_limiter.h | 6 +++ 8 files changed, 126 insertions(+), 71 deletions(-) diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index b444b094a7c62b..db8c6ecaced59a 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -158,18 +158,6 @@ constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID"; constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] = "AWS_CREDENTIALS_PROVIDER_TYPE"; } // namespace -bvar::Adder s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns"); -bvar::Adder s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count"); -bvar::Adder s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count"); -bvar::Adder s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns"); -bvar::Adder s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count"); -bvar::Adder s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count"); - -static std::atomic s3_get_rate_limit_sleep_log_count {0}; -static std::atomic s3_get_rate_limit_rejected_log_count {0}; -static std::atomic s3_put_rate_limit_sleep_log_count {0}; -static std::atomic s3_put_rate_limit_rejected_log_count {0}; - static std::atomic last_s3_get_token_bucket_tokens {0}; static std::atomic last_s3_get_token_limit {0}; static std::atomic last_s3_get_token_per_second {0}; @@ -238,41 +226,8 @@ int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_bur } int64_t apply_s3_rate_limit(S3RateLimitType type) { - auto sleep_duration = S3ClientFactory::instance().rate_limiter(type)->add(1); - int64_t interval = config::s3_rate_limiter_log_interval; - if (interval <= 0 || sleep_duration == 0) { - return sleep_duration; - } - - auto is_get = type == S3RateLimitType::GET; - auto* sleep_log_count = - is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count; - auto* rejected_log_count = - is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count; - int64_t token_per_second = - is_get ? config::s3_get_token_per_second : config::s3_put_token_per_second; - int64_t bucket_tokens = is_get ? config::s3_get_bucket_tokens : config::s3_put_bucket_tokens; - int64_t token_limit = is_get ? config::s3_get_token_limit : config::s3_put_token_limit; - - if (sleep_duration > 0) { - int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1; - if (count == 1 || count % interval == 0) { - LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter" - << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count - << ", token_per_second=" << token_per_second - << ", bucket_tokens=" << bucket_tokens << ", token_limit=" << token_limit; - } - } else { - int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1; - if (count == 1 || count % interval == 0) { - LOG(WARNING) << "S3 " << to_string(type) - << " request is rejected by local rate limiter" - << ", rejected_count=" << count - << ", token_per_second=" << token_per_second - << ", bucket_tokens=" << bucket_tokens << ", token_limit=" << token_limit; - } - } - return sleep_duration; + return doris::apply_s3_rate_limit(type, S3ClientFactory::instance().rate_limiter(type), + config::s3_rate_limiter_log_interval); } S3ClientFactory::S3ClientFactory() { @@ -287,16 +242,11 @@ S3ClientFactory::S3ClientFactory() { _rate_limiters = { std::make_unique( config::s3_get_token_per_second, config::s3_get_bucket_tokens, - config::s3_get_token_limit, - metric_func_factory(s3_get_rate_limit_sleep_ns, - s3_get_rate_limit_sleep_count, - &s3_get_rate_limit_rejected_count)), + config::s3_get_token_limit, s3_rate_limiter_metric_func(S3RateLimitType::GET)), std::make_unique( config::s3_put_token_per_second, config::s3_put_bucket_tokens, config::s3_put_token_limit, - metric_func_factory(s3_put_rate_limit_sleep_ns, - s3_put_rate_limit_sleep_count, - &s3_put_rate_limit_rejected_count))}; + s3_rate_limiter_metric_func(S3RateLimitType::PUT))}; #ifdef USE_AZURE auto azureLogLevel = diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h index cdc8b5cd190891..8fe14abb2ca241 100644 --- a/cloud/src/common/config.h +++ b/cloud/src/common/config.h @@ -254,6 +254,9 @@ CONF_mBool(enable_s3_rate_limiter, "false"); // s3_rate_limit_inject_probility is the probability (0-100) of injecting a rate limit error. CONF_mBool(enable_s3_rate_limit_inject, "false"); CONF_mInt32(s3_rate_limit_inject_probility, "30"); +// Log active S3 rate limiter every N throttled/rejected requests, 0 means no log. +CONF_mInt64(s3_rate_limiter_log_interval, "1000"); +CONF_Validator(s3_rate_limiter_log_interval, [](int64_t config) -> bool { return config >= 0; }); CONF_mInt64(s3_get_bucket_tokens, "1000000000000000000"); CONF_Validator(s3_get_bucket_tokens, [](int64_t config) -> bool { return config > 0; }); diff --git a/cloud/src/recycler/azure_obj_client.cpp b/cloud/src/recycler/azure_obj_client.cpp index 94901cde48b6c1..5390c1c5f99608 100644 --- a/cloud/src/recycler/azure_obj_client.cpp +++ b/cloud/src/recycler/azure_obj_client.cpp @@ -35,6 +35,7 @@ #include "common/config.h" #include "common/logging.h" #include "common/stopwatch.h" +#include "cpp/obj_retry_strategy.h" #include "cpp/sync_point.h" #include "cpp/token_bucket_rate_limiter.h" #include "recycler/s3_accessor.h" @@ -56,7 +57,9 @@ auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) { if (!config::enable_s3_rate_limiter) { return callback(); } - auto sleep_duration = AccessorRateLimiter::instance().rate_limiter(op)->add(1); + auto sleep_duration = + doris::apply_s3_rate_limit(op, AccessorRateLimiter::instance().rate_limiter(op), + config::s3_rate_limiter_log_interval); if (sleep_duration < 0) { throw std::runtime_error("Azure exceeds request limit"); } @@ -81,6 +84,7 @@ ObjectStorageResponse do_azure_client_call(Func f, std::string_view url, std::st try { f(); } catch (Azure::Core::RequestFailedException& e) { + doris::record_object_request_failed(static_cast(e.StatusCode)); auto msg = fmt::format( "Azure request failed because {}, http_code: {}, request_id: {}, url: {}, " "key: {}", @@ -280,6 +284,7 @@ ObjectStorageResponse AzureObjClient::delete_objects(const std::string& bucket, 0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) { continue; } + doris::record_object_request_failed(static_cast(e.StatusCode)); auto msg = fmt::format( "Azure request failed because {}, http code {}, request id {}, url {}", e.Message, static_cast(e.StatusCode), e.RequestId, client_->GetUrl()); @@ -333,4 +338,4 @@ ObjectStorageResponse AzureObjClient::abort_multipart_upload(ObjectStoragePathRe return delete_object(path); } -} // namespace doris::cloud \ No newline at end of file +} // namespace doris::cloud diff --git a/cloud/src/recycler/s3_accessor.cpp b/cloud/src/recycler/s3_accessor.cpp index cc4384b75a504e..d064ef2ef28796 100644 --- a/cloud/src/recycler/s3_accessor.cpp +++ b/cloud/src/recycler/s3_accessor.cpp @@ -73,22 +73,16 @@ bvar::LatencyRecorder s3_get_bucket_version_latency("s3_get_bucket_version"); bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object"); }; // namespace s3_bvar -bvar::Adder get_rate_limit_ns("get_rate_limit_ns"); -bvar::Adder get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num"); -bvar::Adder put_rate_limit_ns("put_rate_limit_ns"); -bvar::Adder put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num"); - AccessorRateLimiter::AccessorRateLimiter() : _rate_limiters( {std::make_unique( config::s3_get_token_per_second, config::s3_get_bucket_tokens, config::s3_get_token_limit, - metric_func_factory(get_rate_limit_ns, get_rate_limit_exceed_req_num)), + s3_rate_limiter_metric_func(S3RateLimitType::GET)), std::make_unique( config::s3_put_token_per_second, config::s3_put_bucket_tokens, config::s3_put_token_limit, - metric_func_factory(put_rate_limit_ns, - put_rate_limit_exceed_req_num))}) {} + s3_rate_limiter_metric_func(S3RateLimitType::PUT))}) {} S3RateLimiterHolder* AccessorRateLimiter::rate_limiter(S3RateLimitType type) { CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type); diff --git a/cloud/src/recycler/s3_obj_client.cpp b/cloud/src/recycler/s3_obj_client.cpp index c45a6aae168c0c..3f299d0fef03e5 100644 --- a/cloud/src/recycler/s3_obj_client.cpp +++ b/cloud/src/recycler/s3_obj_client.cpp @@ -32,6 +32,7 @@ #include "common/config.h" #include "common/logging.h" #include "common/stopwatch.h" +#include "cpp/obj_retry_strategy.h" #include "cpp/sync_point.h" #include "cpp/token_bucket_rate_limiter.h" #include "recycler/s3_accessor.h" @@ -43,6 +44,10 @@ namespace doris::cloud { return {Aws::S3::S3Errors::INTERNAL_FAILURE, "exceeds limit", "exceeds limit", false}; } +void record_s3_request_failed(const Aws::S3::S3Error& error) { + doris::record_object_request_failed(static_cast(error.GetResponseCode())); +} + template auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) { using T = decltype(callback()); @@ -55,7 +60,9 @@ auto s3_rate_limit(S3RateLimitType op, Func callback) -> decltype(callback()) { if (!config::enable_s3_rate_limiter) { return callback(); } - auto sleep_duration = AccessorRateLimiter::instance().rate_limiter(op)->add(1); + auto sleep_duration = + doris::apply_s3_rate_limit(op, AccessorRateLimiter::instance().rate_limiter(op), + config::s3_rate_limiter_log_interval); if (sleep_duration < 0) { return T(s3_error_factory()); } @@ -118,6 +125,7 @@ class S3ObjListIterator final : public ObjectListIterator { return false; } + record_s3_request_failed(outcome.GetError()); LOG_WARNING("failed to list objects") .tag("endpoint", endpoint_) .tag("bucket", req_.GetBucket()) @@ -210,6 +218,7 @@ ObjectStorageResponse S3ObjClient::put_object(ObjectStoragePathRef path, std::st return s3_client_->PutObject(request); }); if (!outcome.IsSuccess()) { + record_s3_request_failed(outcome.GetError()); LOG_WARNING("failed to put object") .tag("endpoint", endpoint_) .tag("bucket", path.bucket) @@ -237,6 +246,7 @@ ObjectStorageResponse S3ObjClient::head_object(ObjectStoragePathRef path, Object } else if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { return 1; } else { + record_s3_request_failed(outcome.GetError()); LOG_WARNING("failed to head object") .tag("endpoint", endpoint_) .tag("bucket", path.bucket) @@ -276,6 +286,7 @@ ObjectStorageResponse S3ObjClient::delete_objects(const std::string& bucket, return s3_client_->DeleteObjects(delete_request); }); if (!delete_outcome.IsSuccess()) { + record_s3_request_failed(delete_outcome.GetError()); LOG_WARNING("failed to delete objects") .tag("endpoint", endpoint_) .tag("bucket", bucket) @@ -326,6 +337,10 @@ ObjectStorageResponse S3ObjClient::delete_object(ObjectStoragePathRef path) { }); TEST_SYNC_POINT_CALLBACK("S3ObjClient::delete_object", &outcome); if (!outcome.IsSuccess()) { + if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { + return {ObjectStorageResponse::NOT_FOUND, outcome.GetError().GetMessage()}; + } + record_s3_request_failed(outcome.GetError()); LOG_WARNING("failed to delete object") .tag("endpoint", endpoint_) .tag("bucket", path.bucket) @@ -334,9 +349,6 @@ ObjectStorageResponse S3ObjClient::delete_object(ObjectStoragePathRef path) { .tag("error", outcome.GetError().GetMessage()) .tag("exception", outcome.GetError().GetExceptionName()) .tag("request_id", outcome.GetError().GetRequestId()); - if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { - return {ObjectStorageResponse::NOT_FOUND, outcome.GetError().GetMessage()}; - } return {ObjectStorageResponse::UNDEFINED, outcome.GetError().GetMessage()}; } return {ObjectStorageResponse::OK}; @@ -365,6 +377,7 @@ ObjectStorageResponse S3ObjClient::get_life_cycle(const std::string& bucket, } } } else { + record_s3_request_failed(outcome.GetError()); LOG_WARNING("Err for check interval: failed to get bucket lifecycle") .tag("endpoint", endpoint_) .tag("bucket", bucket) @@ -397,6 +410,7 @@ ObjectStorageResponse S3ObjClient::check_versioning(const std::string& bucket) { return -1; } } else { + record_s3_request_failed(outcome.GetError()); LOG_WARNING("Err for check interval: failed to get status of bucket versioning") .tag("endpoint", endpoint_) .tag("bucket", bucket) @@ -414,6 +428,10 @@ ObjectStorageResponse S3ObjClient::abort_multipart_upload(ObjectStoragePathRef p request.WithBucket(path.bucket).WithKey(path.key).WithUploadId(upload_id); auto outcome = s3_put_rate_limit([&]() { return s3_client_->AbortMultipartUpload(request); }); if (!outcome.IsSuccess()) { + if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { + return {ObjectStorageResponse::OK}; + } + record_s3_request_failed(outcome.GetError()); LOG_WARNING("failed to abort multipart upload") .tag("endpoint", endpoint_) .tag("bucket", path.bucket) @@ -423,9 +441,6 @@ ObjectStorageResponse S3ObjClient::abort_multipart_upload(ObjectStoragePathRef p .tag("error", outcome.GetError().GetMessage()) .tag("exception", outcome.GetError().GetExceptionName()) .tag("request_id", outcome.GetError().GetRequestId()); - if (outcome.GetError().GetResponseCode() == Aws::Http::HttpResponseCode::NOT_FOUND) { - return {ObjectStorageResponse::OK}; - } return {ObjectStorageResponse::UNDEFINED, outcome.GetError().GetMessage()}; } return {ObjectStorageResponse::OK}; diff --git a/cloud/test/s3_rate_limiter_test.cpp b/cloud/test/s3_rate_limiter_test.cpp index 20496752e586cb..a28bdf3bd2fad4 100644 --- a/cloud/test/s3_rate_limiter_test.cpp +++ b/cloud/test/s3_rate_limiter_test.cpp @@ -29,6 +29,10 @@ using namespace doris::cloud; +namespace doris { +extern bvar::Adder s3_put_rate_limit_rejected_count; +} // namespace doris + int main(int argc, char** argv) { auto conf_file = "doris_cloud.conf"; if (!doris::cloud::config::init(conf_file, true)) { @@ -131,3 +135,19 @@ TEST(S3RateLimiterHolderTest, BvarMetric) { EXPECT_EQ(sleep_time, -1); EXPECT_EQ(rate_limit_rejected_count.get_value(), 1); } + +TEST(S3RateLimiterHolderTest, ApplyS3RateLimitRecordsRejectedMetric) { + auto rate_limiter_holder = doris::S3RateLimiterHolder( + 125, 250, 1, doris::s3_rate_limiter_metric_func(doris::S3RateLimitType::PUT)); + auto rejected_count = doris::s3_put_rate_limit_rejected_count.get_value(); + + int64_t sleep_time = doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, + &rate_limiter_holder, 1); + EXPECT_EQ(sleep_time, 0); + EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count); + + sleep_time = + doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1); + EXPECT_EQ(sleep_time, -1); + EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count + 1); +} diff --git a/common/cpp/token_bucket_rate_limiter.cpp b/common/cpp/token_bucket_rate_limiter.cpp index 0e30f870de368e..f3480394b7fd98 100644 --- a/common/cpp/token_bucket_rate_limiter.cpp +++ b/common/cpp/token_bucket_rate_limiter.cpp @@ -20,6 +20,7 @@ #include #include // IWYU pragma: export +#include #include #include #include @@ -32,6 +33,18 @@ namespace doris { // Just 10^6. static constexpr auto NS = 1000000000UL; +bvar::Adder s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns"); +bvar::Adder s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count"); +bvar::Adder s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count"); +bvar::Adder s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns"); +bvar::Adder s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count"); +bvar::Adder s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count"); + +static std::atomic s3_get_rate_limit_sleep_log_count {0}; +static std::atomic s3_get_rate_limit_rejected_log_count {0}; +static std::atomic s3_put_rate_limit_sleep_log_count {0}; +static std::atomic s3_put_rate_limit_rejected_log_count {0}; + class TokenBucketRateLimiter::SimpleSpinLock { public: SimpleSpinLock() = default; @@ -154,4 +167,53 @@ S3RateLimitType string_to_s3_rate_limit_type(std::string_view value) { } return S3RateLimitType::UNKNOWN; } + +std::function s3_rate_limiter_metric_func(S3RateLimitType type) { + switch (type) { + case S3RateLimitType::GET: + return metric_func_factory(s3_get_rate_limit_sleep_ns, s3_get_rate_limit_sleep_count, + &s3_get_rate_limit_rejected_count); + case S3RateLimitType::PUT: + return metric_func_factory(s3_put_rate_limit_sleep_ns, s3_put_rate_limit_sleep_count, + &s3_put_rate_limit_rejected_count); + default: + return [](int64_t) {}; + } +} + +int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter, + int64_t log_interval) { + auto sleep_duration = rate_limiter->add(1); + if (log_interval <= 0 || sleep_duration == 0) { + return sleep_duration; + } + + auto is_get = type == S3RateLimitType::GET; + auto* sleep_log_count = + is_get ? &s3_get_rate_limit_sleep_log_count : &s3_put_rate_limit_sleep_log_count; + auto* rejected_log_count = + is_get ? &s3_get_rate_limit_rejected_log_count : &s3_put_rate_limit_rejected_log_count; + + if (sleep_duration > 0) { + int64_t count = sleep_log_count->fetch_add(1, std::memory_order_relaxed) + 1; + if (count == 1 || count % log_interval == 0) { + LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter" + << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count + << ", token_per_second=" << rate_limiter->get_max_speed() + << ", bucket_tokens=" << rate_limiter->get_max_burst() + << ", token_limit=" << rate_limiter->get_limit(); + } + } else { + int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1; + if (count == 1 || count % log_interval == 0) { + LOG(WARNING) << "S3 " << to_string(type) + << " request is rejected by local rate limiter" + << ", rejected_count=" << count + << ", token_per_second=" << rate_limiter->get_max_speed() + << ", bucket_tokens=" << rate_limiter->get_max_burst() + << ", token_limit=" << rate_limiter->get_limit(); + } + } + return sleep_duration; +} } // namespace doris diff --git a/common/cpp/token_bucket_rate_limiter.h b/common/cpp/token_bucket_rate_limiter.h index a2f571dc1fa9e0..876f5d4fb3ecbc 100644 --- a/common/cpp/token_bucket_rate_limiter.h +++ b/common/cpp/token_bucket_rate_limiter.h @@ -21,6 +21,8 @@ #include #include #include +#include +#include namespace doris { enum class S3RateLimitType : int { @@ -100,4 +102,8 @@ class TokenBucketRateLimiterHolder { using S3RateLimiter = TokenBucketRateLimiter; using S3RateLimiterHolder = TokenBucketRateLimiterHolder; + +std::function s3_rate_limiter_metric_func(S3RateLimitType type); +int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter, + int64_t log_interval); } // namespace doris From e0620e17ce0fd50225951e7710791b7854f494ef Mon Sep 17 00:00:00 2001 From: gavinchou Date: Fri, 5 Jun 2026 11:45:46 +0800 Subject: [PATCH 3/4] fix format --- be/src/util/s3_util.cpp | 3 +-- cloud/src/recycler/s3_accessor.cpp | 17 ++++++++--------- cloud/test/s3_rate_limiter_test.cpp | 7 +++---- 3 files changed, 12 insertions(+), 15 deletions(-) diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp index db8c6ecaced59a..9c8a073438d09e 100644 --- a/be/src/util/s3_util.cpp +++ b/be/src/util/s3_util.cpp @@ -245,8 +245,7 @@ S3ClientFactory::S3ClientFactory() { config::s3_get_token_limit, s3_rate_limiter_metric_func(S3RateLimitType::GET)), std::make_unique( config::s3_put_token_per_second, config::s3_put_bucket_tokens, - config::s3_put_token_limit, - s3_rate_limiter_metric_func(S3RateLimitType::PUT))}; + config::s3_put_token_limit, s3_rate_limiter_metric_func(S3RateLimitType::PUT))}; #ifdef USE_AZURE auto azureLogLevel = diff --git a/cloud/src/recycler/s3_accessor.cpp b/cloud/src/recycler/s3_accessor.cpp index d064ef2ef28796..b68f1eb3cb4e41 100644 --- a/cloud/src/recycler/s3_accessor.cpp +++ b/cloud/src/recycler/s3_accessor.cpp @@ -74,15 +74,14 @@ bvar::LatencyRecorder s3_copy_object_latency("s3_copy_object"); }; // namespace s3_bvar AccessorRateLimiter::AccessorRateLimiter() - : _rate_limiters( - {std::make_unique( - config::s3_get_token_per_second, config::s3_get_bucket_tokens, - config::s3_get_token_limit, - s3_rate_limiter_metric_func(S3RateLimitType::GET)), - std::make_unique( - config::s3_put_token_per_second, config::s3_put_bucket_tokens, - config::s3_put_token_limit, - s3_rate_limiter_metric_func(S3RateLimitType::PUT))}) {} + : _rate_limiters({std::make_unique( + config::s3_get_token_per_second, config::s3_get_bucket_tokens, + config::s3_get_token_limit, + s3_rate_limiter_metric_func(S3RateLimitType::GET)), + std::make_unique( + config::s3_put_token_per_second, config::s3_put_bucket_tokens, + config::s3_put_token_limit, + s3_rate_limiter_metric_func(S3RateLimitType::PUT))}) {} S3RateLimiterHolder* AccessorRateLimiter::rate_limiter(S3RateLimitType type) { CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type); diff --git a/cloud/test/s3_rate_limiter_test.cpp b/cloud/test/s3_rate_limiter_test.cpp index a28bdf3bd2fad4..149d43cc8d13aa 100644 --- a/cloud/test/s3_rate_limiter_test.cpp +++ b/cloud/test/s3_rate_limiter_test.cpp @@ -141,13 +141,12 @@ TEST(S3RateLimiterHolderTest, ApplyS3RateLimitRecordsRejectedMetric) { 125, 250, 1, doris::s3_rate_limiter_metric_func(doris::S3RateLimitType::PUT)); auto rejected_count = doris::s3_put_rate_limit_rejected_count.get_value(); - int64_t sleep_time = doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, - &rate_limiter_holder, 1); + int64_t sleep_time = + doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1); EXPECT_EQ(sleep_time, 0); EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count); - sleep_time = - doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1); + sleep_time = doris::apply_s3_rate_limit(doris::S3RateLimitType::PUT, &rate_limiter_holder, 1); EXPECT_EQ(sleep_time, -1); EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count + 1); } From 366f14b1f677ecdc34d0defec59dbe72b8b352f9 Mon Sep 17 00:00:00 2001 From: gavinchou Date: Sat, 11 Jul 2026 20:45:35 +0800 Subject: [PATCH 4/4] [fix](s3) Avoid race in rate limiter logging --- be/src/io/fs/azure_obj_storage_client.cpp | 1 + cloud/test/s3_rate_limiter_test.cpp | 37 ++++++++++++++++++ common/cpp/token_bucket_rate_limiter.cpp | 47 ++++++++++++++++------- common/cpp/token_bucket_rate_limiter.h | 18 ++++++--- 4 files changed, 84 insertions(+), 19 deletions(-) diff --git a/be/src/io/fs/azure_obj_storage_client.cpp b/be/src/io/fs/azure_obj_storage_client.cpp index 8b09a2a4aafd11..4ac9117fad9ae3 100644 --- a/be/src/io/fs/azure_obj_storage_client.cpp +++ b/be/src/io/fs/azure_obj_storage_client.cpp @@ -188,6 +188,7 @@ struct AzureBatchDeleter { 0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) { continue; } + doris::record_object_request_failed(static_cast(e.StatusCode)); auto msg = fmt::format( "Azure request failed because {}, error msg {}, http code {}, path msg " "{}{}", diff --git a/cloud/test/s3_rate_limiter_test.cpp b/cloud/test/s3_rate_limiter_test.cpp index 149d43cc8d13aa..05fbb2ce35ec03 100644 --- a/cloud/test/s3_rate_limiter_test.cpp +++ b/cloud/test/s3_rate_limiter_test.cpp @@ -150,3 +150,40 @@ TEST(S3RateLimiterHolderTest, ApplyS3RateLimitRecordsRejectedMetric) { EXPECT_EQ(sleep_time, -1); EXPECT_EQ(doris::s3_put_rate_limit_rejected_count.get_value(), rejected_count + 1); } + +TEST(S3RateLimiterHolderTest, ConcurrentResetReturnsConsistentConfig) { + constexpr size_t config_a_speed = 101; + constexpr size_t config_a_burst = 102; + constexpr size_t config_a_limit = 103; + constexpr size_t config_b_speed = 201; + constexpr size_t config_b_burst = 202; + constexpr size_t config_b_limit = 203; + + doris::S3RateLimiterHolder rate_limiter_holder(config_a_speed, config_a_burst, config_a_limit, + [](int64_t) {}); + std::atomic start {false}; + std::thread reset_thread([&]() { + while (!start.load(std::memory_order_acquire)) { + std::this_thread::yield(); + } + for (size_t i = 0; i < 10000; ++i) { + if (i % 2 == 0) { + rate_limiter_holder.reset(config_b_speed, config_b_burst, config_b_limit); + } else { + rate_limiter_holder.reset(config_a_speed, config_a_burst, config_a_limit); + } + } + }); + + start.store(true, std::memory_order_release); + for (size_t i = 0; i < 10000; ++i) { + auto result = rate_limiter_holder.add_with_config(0); + bool is_config_a = result.max_speed == config_a_speed && + result.max_burst == config_a_burst && result.limit == config_a_limit; + bool is_config_b = result.max_speed == config_b_speed && + result.max_burst == config_b_burst && result.limit == config_b_limit; + EXPECT_TRUE(is_config_a || is_config_b); + } + + reset_thread.join(); +} diff --git a/common/cpp/token_bucket_rate_limiter.cpp b/common/cpp/token_bucket_rate_limiter.cpp index f3480394b7fd98..128501ae7cf469 100644 --- a/common/cpp/token_bucket_rate_limiter.cpp +++ b/common/cpp/token_bucket_rate_limiter.cpp @@ -131,13 +131,20 @@ TokenBucketRateLimiterHolder::TokenBucketRateLimiterHolder(size_t max_speed, siz metric_func(std::move(metric_func)) {} int64_t TokenBucketRateLimiterHolder::add(size_t amount) { - int64_t sleep; + return add_with_config(amount).sleep_duration; +} + +TokenBucketRateLimiterResult TokenBucketRateLimiterHolder::add_with_config(size_t amount) { + TokenBucketRateLimiterResult result; { std::shared_lock read {rate_limiter_rw_lock}; - sleep = rate_limiter->add(amount); + result = {.sleep_duration = rate_limiter->add(amount), + .max_speed = rate_limiter->get_max_speed(), + .max_burst = rate_limiter->get_max_burst(), + .limit = rate_limiter->get_limit()}; } - metric_func(sleep); - return sleep; + metric_func(result.sleep_duration); + return result; } int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size_t limit) { @@ -148,6 +155,21 @@ int TokenBucketRateLimiterHolder::reset(size_t max_speed, size_t max_burst, size return 0; } +size_t TokenBucketRateLimiterHolder::get_max_speed() const { + std::shared_lock read {rate_limiter_rw_lock}; + return rate_limiter->get_max_speed(); +} + +size_t TokenBucketRateLimiterHolder::get_max_burst() const { + std::shared_lock read {rate_limiter_rw_lock}; + return rate_limiter->get_max_burst(); +} + +size_t TokenBucketRateLimiterHolder::get_limit() const { + std::shared_lock read {rate_limiter_rw_lock}; + return rate_limiter->get_limit(); +} + std::string to_string(S3RateLimitType type) { switch (type) { case S3RateLimitType::GET: @@ -183,7 +205,8 @@ std::function s3_rate_limiter_metric_func(S3RateLimitType type) { int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limiter, int64_t log_interval) { - auto sleep_duration = rate_limiter->add(1); + auto result = rate_limiter->add_with_config(1); + auto sleep_duration = result.sleep_duration; if (log_interval <= 0 || sleep_duration == 0) { return sleep_duration; } @@ -199,19 +222,17 @@ int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder* rate_limi if (count == 1 || count % log_interval == 0) { LOG(INFO) << "S3 " << to_string(type) << " request is throttled by local rate limiter" << ", sleep_ms=" << sleep_duration / 1000000 << ", sleep_count=" << count - << ", token_per_second=" << rate_limiter->get_max_speed() - << ", bucket_tokens=" << rate_limiter->get_max_burst() - << ", token_limit=" << rate_limiter->get_limit(); + << ", token_per_second=" << result.max_speed + << ", bucket_tokens=" << result.max_burst << ", token_limit=" << result.limit; } } else { int64_t count = rejected_log_count->fetch_add(1, std::memory_order_relaxed) + 1; if (count == 1 || count % log_interval == 0) { - LOG(WARNING) << "S3 " << to_string(type) - << " request is rejected by local rate limiter" + LOG(WARNING) << "S3 " << to_string(type) << " request is rejected by local rate limiter" << ", rejected_count=" << count - << ", token_per_second=" << rate_limiter->get_max_speed() - << ", bucket_tokens=" << rate_limiter->get_max_burst() - << ", token_limit=" << rate_limiter->get_limit(); + << ", token_per_second=" << result.max_speed + << ", bucket_tokens=" << result.max_burst + << ", token_limit=" << result.limit; } } return sleep_duration; diff --git a/common/cpp/token_bucket_rate_limiter.h b/common/cpp/token_bucket_rate_limiter.h index 876f5d4fb3ecbc..57f9205527acc6 100644 --- a/common/cpp/token_bucket_rate_limiter.h +++ b/common/cpp/token_bucket_rate_limiter.h @@ -77,6 +77,13 @@ class TokenBucketRateLimiter { long _prev_ns_count {0}; // Previous `add` call time (in nanoseconds). }; +struct TokenBucketRateLimiterResult { + int64_t sleep_duration; + size_t max_speed; + size_t max_burst; + size_t limit; +}; + class TokenBucketRateLimiterHolder { public: TokenBucketRateLimiterHolder(size_t max_speed, size_t max_burst, size_t limit, @@ -84,17 +91,16 @@ class TokenBucketRateLimiterHolder { ~TokenBucketRateLimiterHolder(); int64_t add(size_t amount); + TokenBucketRateLimiterResult add_with_config(size_t amount); int reset(size_t max_speed, size_t max_burst, size_t limit); - size_t get_max_speed() const { return rate_limiter->get_max_speed(); } - - size_t get_max_burst() const { return rate_limiter->get_max_burst(); } - - size_t get_limit() const { return rate_limiter->get_limit(); } + size_t get_max_speed() const; + size_t get_max_burst() const; + size_t get_limit() const; private: - std::shared_mutex rate_limiter_rw_lock; + mutable std::shared_mutex rate_limiter_rw_lock; std::unique_ptr rate_limiter; // Record the correspoding sleeping time(unit is ms) std::function metric_func;