[fix](s3) Add S3 rate limit observability

be/src/common/config.cpp

@@ -1523,6 +1523,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");
 

be/src/common/config.h

@@ -1604,6 +1604,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

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 "
                         "{}{}",

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
+} // namespace doris::io

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 =

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();

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; });
 

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
+} // namespace doris::cloud

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);