This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 68938c5ca31 [fix](s3) Add S3 rate limit observability (#64038)
68938c5ca31 is described below
commit 68938c5ca31646686fffcb1e2dd153567539db47
Author: Gavin Chou <[email protected]>
AuthorDate: Wed Jul 15 21:51:38 2026 +0800
[fix](s3) Add S3 rate limit observability (#64038)
### What problem does this PR solve?
Add observability for S3 local rate limiting and S3 429 responses so
throttling-related performance issues can be diagnosed and alerted more
directly.
### What is changed?
- Add explicit S3 local rate limiter bvars for sleep duration, sleep
count, and rejected count.
- Rename the old ambiguous S3 local rate limiter bvars so their meaning
is clear.
- Log the first and every N local S3 rate limiter throttled/rejected
requests without reading bvar values for decisions.
- Add S3 429 retry/failure bvars for S3/Azure object clients.
- Update S3 rate limiter unit coverage for sleep and rejected metrics.
### New or renamed bvar metrics
Renamed local S3 GET limiter metrics:
- `get_rate_limit_ns` -> `s3_get_rate_limit_sleep_ns`
- `get_rate_limit_exceed_req_num` -> `s3_get_rate_limit_sleep_count`
Renamed local S3 PUT limiter metrics:
- `put_rate_limit_ns` -> `s3_put_rate_limit_sleep_ns`
- `put_rate_limit_exceed_req_num` -> `s3_put_rate_limit_sleep_count`
New local S3 limiter rejection metrics:
- `s3_get_rate_limit_rejected_count`
- `s3_put_rate_limit_rejected_count`
New S3 429 metrics:
- `s3_request_retry_too_many_requests_count`
- `s3_request_failed_too_many_requests_count`
Metric examples:
```text
s3_get_rate_limit_sleep_ns : 123456789
s3_get_rate_limit_sleep_count : 42
s3_get_rate_limit_rejected_count : 3
s3_put_rate_limit_sleep_ns : 987654321
s3_put_rate_limit_sleep_count : 27
s3_put_rate_limit_rejected_count : 1
s3_request_retry_too_many_requests_count : 8
s3_request_failed_too_many_requests_count : 2
```
Example local limiter logs:
```text
S3 GET request is throttled by local rate limiter, sleep_ms=12,
sleep_count=1000, token_per_second=1000, bucket_tokens=2000, token_limit=5000
S3 PUT request is rejected by local rate limiter, rejected_count=1,
token_per_second=1000, bucket_tokens=2000, token_limit=5000
```
### Tests
- `sh run-cloud-ut.sh --run --filter=s3_rate_limiter_test:*`
- `sh run-be-ut.sh --run --filter=S3FileWriterTest.*`
Note: an earlier BE run used an incorrect lowercase filter
(`s3_file_writer_test:*`), which matched no suite and started running
the full BE suite; it was stopped and replaced by the correct
`S3FileWriterTest.*` run above.
---------
Co-authored-by: gavinchou <[email protected]>
---
be/src/common/config.cpp | 3 +
be/src/common/config.h | 1 +
be/src/io/fs/azure_obj_storage_client.cpp | 5 +-
be/src/io/fs/s3_obj_storage_client.cpp | 20 ++++++-
be/src/util/s3_util.cpp | 16 +++---
be/src/util/s3_util.h | 1 +
cloud/src/common/config.h | 3 +
cloud/src/recycler/azure_obj_client.cpp | 9 ++-
cloud/src/recycler/s3_accessor.cpp | 23 +++-----
cloud/src/recycler/s3_obj_client.cpp | 29 +++++++---
cloud/test/s3_rate_limiter_test.cpp | 81 ++++++++++++++++++++++++---
common/cpp/obj_retry_strategy.cpp | 17 ++++++
common/cpp/obj_retry_strategy.h | 2 +
common/cpp/token_bucket_rate_limiter.cpp | 91 +++++++++++++++++++++++++++++--
common/cpp/token_bucket_rate_limiter.h | 38 +++++++++----
15 files changed, 280 insertions(+), 59 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index ca20cb63990..b2d51ecbb6d 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1539,6 +1539,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 2a6b21ccf8c..25c7a8ef89d 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1617,6 +1617,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 85b81c1deb2..4ac9117fad9 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<int>(e.StatusCode));
auto tls_debug_suffix = build_azure_tls_debug_suffix(
fmt::format("{} {}", e.what(), e.Message), tls_debug_context);
auto msg = fmt::format(
@@ -186,6 +188,7 @@ struct AzureBatchDeleter {
0 == strcmp(e.ErrorCode.c_str(), BlobNotFound)) {
continue;
}
+
doris::record_object_request_failed(static_cast<int>(e.StatusCode));
auto msg = fmt::format(
"Azure request failed because {}, error msg {}, http
code {}, path msg "
"{}{}",
diff --git a/be/src/io/fs/s3_obj_storage_client.cpp
b/be/src/io/fs/s3_obj_storage_client.cpp
index 1b8bf7b473c..f9ed8e155ff 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 <typename Func>
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<int>(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<ErrorCode::NOT_FOUND, false>(""))}};
} 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<int>(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<int>(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<int>(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 2021b80fda6..c6e86a7f4b2 100644
--- a/be/src/util/s3_util.cpp
+++ b/be/src/util/s3_util.cpp
@@ -158,11 +158,6 @@ constexpr char S3_EXTERNAL_ID[] = "AWS_EXTERNAL_ID";
constexpr char S3_CREDENTIALS_PROVIDER_TYPE[] =
"AWS_CREDENTIALS_PROVIDER_TYPE";
} // namespace
-bvar::Adder<int64_t> get_rate_limit_ns("get_rate_limit_ns");
-bvar::Adder<int64_t>
get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num");
-bvar::Adder<int64_t> put_rate_limit_ns("put_rate_limit_ns");
-bvar::Adder<int64_t>
put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num");
-
static std::atomic<int64_t> last_s3_get_token_bucket_tokens {0};
static std::atomic<int64_t> last_s3_get_token_limit {0};
static std::atomic<int64_t> last_s3_get_token_per_second {0};
@@ -230,6 +225,11 @@ 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) {
+ return doris::apply_s3_rate_limit(type,
S3ClientFactory::instance().rate_limiter(type),
+ config::s3_rate_limiter_log_interval);
+}
+
S3ClientFactory::S3ClientFactory() {
_aws_options = Aws::SDKOptions {};
auto logLevel =
static_cast<Aws::Utils::Logging::LogLevel>(config::aws_log_level);
@@ -242,12 +242,10 @@ S3ClientFactory::S3ClientFactory() {
_rate_limiters = {
std::make_unique<S3RateLimiterHolder>(
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)),
+ config::s3_get_token_limit,
s3_rate_limiter_metric_func(S3RateLimitType::GET)),
std::make_unique<S3RateLimiterHolder>(
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))};
+ config::s3_put_token_limit,
s3_rate_limiter_metric_func(S3RateLimitType::PUT))};
#ifdef USE_AZURE
auto azureLogLevel =
diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h
index 5546db857c9..064f88accd8 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);
// Rebuild the S3 GET/PUT rate limiters if the related configs have changed.
// Safe to call periodically; it is a no-op when nothing changed.
void check_s3_rate_limiter_config_changed();
diff --git a/cloud/src/common/config.h b/cloud/src/common/config.h
index 926a73a2c4c..a4a2c17441b 100644
--- a/cloud/src/common/config.h
+++ b/cloud/src/common/config.h
@@ -282,6 +282,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 94901cde48b..5390c1c5f99 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<int>(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<int>(e.StatusCode));
auto msg = fmt::format(
"Azure request failed because {}, http code {},
request id {}, url {}",
e.Message, static_cast<int>(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 8ff29694bd7..8fb85d54a2c 100644
--- a/cloud/src/recycler/s3_accessor.cpp
+++ b/cloud/src/recycler/s3_accessor.cpp
@@ -73,22 +73,15 @@ 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<int64_t> get_rate_limit_ns("get_rate_limit_ns");
-bvar::Adder<int64_t>
get_rate_limit_exceed_req_num("get_rate_limit_exceed_req_num");
-bvar::Adder<int64_t> put_rate_limit_ns("put_rate_limit_ns");
-bvar::Adder<int64_t>
put_rate_limit_exceed_req_num("put_rate_limit_exceed_req_num");
-
AccessorRateLimiter::AccessorRateLimiter()
- : _rate_limiters(
- {std::make_unique<S3RateLimiterHolder>(
- 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)),
- std::make_unique<S3RateLimiterHolder>(
- 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))}) {}
+ : _rate_limiters({std::make_unique<S3RateLimiterHolder>(
+ 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<S3RateLimiterHolder>(
+ 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/src/recycler/s3_obj_client.cpp
b/cloud/src/recycler/s3_obj_client.cpp
index c45a6aae168..3f299d0fef0 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<int>(error.GetResponseCode()));
+}
+
template <typename Func>
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 @@ public:
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 b1fd5a8115f..05fbb2ce35e 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<int64_t> 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)) {
@@ -107,18 +111,79 @@ TEST(S3RateLimiterTest, ExceedLimit) {
}
TEST(S3RateLimiterHolderTest, BvarMetric) {
- bvar::Adder<int64_t> rate_limit_ns("rate_limit_ns");
- bvar::Adder<int64_t>
rate_limit_exceed_req_num("rate_limit_exceed_req_num");
+ bvar::Adder<int64_t> rate_limit_sleep_ns("rate_limit_sleep_ns");
+ bvar::Adder<int64_t> rate_limit_sleep_count("rate_limit_sleep_count");
+ bvar::Adder<int64_t>
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);
+}
+
+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);
+}
+
+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<bool> 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/obj_retry_strategy.cpp
b/common/cpp/obj_retry_strategy.cpp
index 58ca47f6a68..83992255364 100644
--- a/common/cpp/obj_retry_strategy.cpp
+++ b/common/cpp/obj_retry_strategy.cpp
@@ -24,6 +24,16 @@
namespace doris {
bvar::Adder<int64_t> object_request_retry_count("object_request_retry_count");
+bvar::Adder<int64_t> s3_request_retry_too_many_requests_count(
+ "s3_request_retry_too_many_requests_count");
+bvar::Adder<int64_t> 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<int>(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) {
+ s3_request_failed_too_many_requests_count << 1;
+ }
+}
S3CustomRetryStrategy::S3CustomRetryStrategy(int maxRetries, bool
retry_slow_down)
: DefaultRetryStrategy(maxRetries), _retry_slow_down(retry_slow_down)
{}
@@ -43,6 +53,9 @@ bool S3CustomRetryStrategy::ShouldRetry(const
Aws::Client::AWSError<Aws::Client:
if (Aws::Http::IsRetryableHttpResponseCode(error.GetResponseCode()) ||
error.ShouldRetry()) {
object_request_retry_count << 1;
+ if (error.GetResponseCode() ==
Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS) {
+ s3_request_retry_too_many_requests_count << 1;
+ }
LOG(INFO) << "retry due to error: " << error << ", attempt: " <<
attemptedRetries + 1 << "/"
<< m_maxRetries;
return true;
@@ -65,6 +78,10 @@ std::unique_ptr<Azure::Core::Http::RawResponse>
AzureRetryRecordPolicy::Send(
static_cast<int>(response->GetStatusCode()) < 200) {
if (retry_count > 0) {
object_request_retry_count << 1;
+ if (static_cast<int>(response->GetStatusCode()) ==
+
static_cast<int>(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.
diff --git a/common/cpp/obj_retry_strategy.h b/common/cpp/obj_retry_strategy.h
index aaf35f6b100..2eafe114564 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, bool retry_slow_down = true);
diff --git a/common/cpp/token_bucket_rate_limiter.cpp
b/common/cpp/token_bucket_rate_limiter.cpp
index 0e30f870de3..128501ae7cf 100644
--- a/common/cpp/token_bucket_rate_limiter.cpp
+++ b/common/cpp/token_bucket_rate_limiter.cpp
@@ -20,6 +20,7 @@
#include <bthread/bthread.h>
#include <glog/logging.h> // IWYU pragma: export
+#include <atomic>
#include <chrono>
#include <mutex>
#include <thread>
@@ -32,6 +33,18 @@ namespace doris {
// Just 10^6.
static constexpr auto NS = 1000000000UL;
+bvar::Adder<int64_t> s3_get_rate_limit_sleep_ns("s3_get_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_get_rate_limit_sleep_count("s3_get_rate_limit_sleep_count");
+bvar::Adder<int64_t>
s3_get_rate_limit_rejected_count("s3_get_rate_limit_rejected_count");
+bvar::Adder<int64_t> s3_put_rate_limit_sleep_ns("s3_put_rate_limit_sleep_ns");
+bvar::Adder<int64_t>
s3_put_rate_limit_sleep_count("s3_put_rate_limit_sleep_count");
+bvar::Adder<int64_t>
s3_put_rate_limit_rejected_count("s3_put_rate_limit_rejected_count");
+
+static std::atomic<int64_t> s3_get_rate_limit_sleep_log_count {0};
+static std::atomic<int64_t> s3_get_rate_limit_rejected_log_count {0};
+static std::atomic<int64_t> s3_put_rate_limit_sleep_log_count {0};
+static std::atomic<int64_t> s3_put_rate_limit_rejected_log_count {0};
+
class TokenBucketRateLimiter::SimpleSpinLock {
public:
SimpleSpinLock() = default;
@@ -118,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) {
@@ -135,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:
@@ -154,4 +189,52 @@ S3RateLimitType
string_to_s3_rate_limit_type(std::string_view value) {
}
return S3RateLimitType::UNKNOWN;
}
+
+std::function<void(int64_t)> 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 result = rate_limiter->add_with_config(1);
+ auto sleep_duration = result.sleep_duration;
+ 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=" << 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"
+ << ", rejected_count=" << count
+ << ", token_per_second=" << result.max_speed
+ << ", bucket_tokens=" << result.max_burst
+ << ", token_limit=" << result.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 dbeca3b897d..57f9205527a 100644
--- a/common/cpp/token_bucket_rate_limiter.h
+++ b/common/cpp/token_bucket_rate_limiter.h
@@ -21,6 +21,8 @@
#include <functional>
#include <memory>
#include <shared_mutex>
+#include <string>
+#include <string_view>
namespace doris {
enum class S3RateLimitType : int {
@@ -32,11 +34,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<int64_t>& ns_bvar,
bvar::Adder<int64_t>& req_num_bvar) {
- return [&](int64_t ns) {
+inline auto metric_func_factory(bvar::Adder<int64_t>& sleep_ns_bvar,
+ bvar::Adder<int64_t>& sleep_count_bvar,
+ bvar::Adder<int64_t>* 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;
}
};
}
@@ -71,6 +77,13 @@ private:
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,
@@ -78,17 +91,16 @@ public:
~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<TokenBucketRateLimiter> rate_limiter;
// Record the correspoding sleeping time(unit is ms)
std::function<void(int64_t)> metric_func;
@@ -96,4 +108,8 @@ private:
using S3RateLimiter = TokenBucketRateLimiter;
using S3RateLimiterHolder = TokenBucketRateLimiterHolder;
-} // namespace doris
\ No newline at end of file
+
+std::function<void(int64_t)> s3_rate_limiter_metric_func(S3RateLimitType type);
+int64_t apply_s3_rate_limit(S3RateLimitType type, S3RateLimiterHolder*
rate_limiter,
+ int64_t log_interval);
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
