github-actions[bot] commented on code in PR #65420: URL: https://github.com/apache/doris/pull/65420#discussion_r3611810685
########## be/src/io/fs/rate_limited_obj_storage_client.cpp: ########## @@ -0,0 +1,139 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "io/fs/rate_limited_obj_storage_client.h" + +#include "common/status.h" +#include "util/s3_rate_limiter_manager.h" + +namespace doris::io { +namespace { + +ObjectStorageResponse rate_limited_response(S3RateLimitType type) { + return {.status = convert_to_obj_response(Status::Error<ErrorCode::EXCEEDED_LIMIT, false>( + "s3 {} request exceeds request limit, rejected by BE rate limiter", + to_string(type))), + .http_code = 429}; Review Comment: [P2] Do not retry a permanent local count-limit rejection `guard.ok()` is false only when the legacy cumulative `token_limit` has already been exceeded; ordinary QPS throttling sleeps and still returns success. Labeling that permanent local failure as HTTP 429 makes `S3FileReader::read_at_impl()` treat it as a transient provider throttle and repeat the admission/backoff loop through `max_s3_client_retry`, although every retry only increments the exhausted count again. The removed S3-local path returned a non-retryable failure. Please preserve a non-retryable discriminator (or have the reader recognize `EXCEEDED_LIMIT`) and add a reader-level test showing one immediate failure without 429 backoff. ########## common/cpp/token_bucket_rate_limiter.cpp: ########## @@ -124,33 +124,58 @@ int64_t TokenBucketRateLimiter::add(size_t amount) { return sleep_time_ns; } +void TokenBucketRateLimiter::refund(size_t amount) { Review Comment: [P2] Apply refill before debiting or refunding `_update_remain_token()` caps only after subtracting the new amount. With rate/burst `B`, drain to zero, idle 2 seconds, then `add(B)` leaves `min(0 + 2B - B, B) = B`, so an immediate second `B` also passes. `refund()` adds a second ordering error by returning tokens without first advancing `_prev_ns_count`, so the next add refills an already-refunded balance. Please cap accrued refill before an admission debit; at refund time, advance/cap refill and the clock before adding/capping the returned tokens. Add controllable-clock tests for post-idle admission and delayed short-read refund. ########## be/src/io/fs/rate_limited_obj_storage_client.h: ########## @@ -0,0 +1,70 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#pragma once + +#include <memory> + +#include "io/fs/obj_storage_client.h" + +namespace doris::io { + +// Decorator that applies the process-wide S3 GET/PUT QPS and bandwidth rate limiters +// in front of any ObjStorageClient. This is the single place where rate limiting is +// wired into the object storage path: provider clients (S3, Azure, future GCP, ...) +// contain no rate limiting code, and S3ClientFactory decides at construction time +// whether to wrap a client (internal storage-vault buckets) or return it bare +// (external buckets: S3 load, TVF, external catalogs in cloud mode). +// +// Each public API call is charged once against the QPS bucket, and data-carrying +// calls additionally reserve their payload size from the bytes bucket (reconciled +// with the actually transferred size for reads). Note that APIs which internally +// paginate (list_objects, delete_objects_recursively) are charged once per logical Review Comment: [P1] Keep QPS admission at each provider request boundary The legacy fallback is described as bit-for-bit compatible, but this moves it from provider requests to logical Doris calls. One admitted `list_objects()` can issue every S3/Azure page back-to-back, and `delete_objects_recursively()` consumes one PUT token while each page performs both a LIST/GET and a delete request; Azure batching has the same multiplication. A 1-QPS setting can therefore send an unbounded service burst, while Azure's no-op multipart create consumes a token despite sending nothing. This is distinct from the existing SDK-retry byte thread because it occurs on successful page/batch paths and bypasses the GET bucket entirely. Please charge every actual page/batch with the matching GET/PUT limiter (before signing) and test provider call counts. ########## be/src/util/s3_rate_limiter_manager.cpp: ########## @@ -0,0 +1,204 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "util/s3_rate_limiter_manager.h" + +#include <algorithm> +#include <limits> +#include <thread> + +#include "common/config.h" +#include "common/logging.h" +#include "util/cgroup_util.h" + +namespace doris { + +bvar::Adder<int64_t> s3_get_bytes_rate_limit_sleep_ns("s3_get_bytes_rate_limit_sleep_ns"); +bvar::Adder<int64_t> s3_get_bytes_rate_limit_sleep_count("s3_get_bytes_rate_limit_sleep_count"); +bvar::Adder<int64_t> s3_put_bytes_rate_limit_sleep_ns("s3_put_bytes_rate_limit_sleep_ns"); +bvar::Adder<int64_t> s3_put_bytes_rate_limit_sleep_count("s3_put_bytes_rate_limit_sleep_count"); + +namespace { + +std::function<void(int64_t)> bytes_rate_limiter_metric_func(S3RateLimitType type) { + switch (type) { + case S3RateLimitType::GET: + return metric_func_factory(s3_get_bytes_rate_limit_sleep_ns, + s3_get_bytes_rate_limit_sleep_count); + case S3RateLimitType::PUT: + return metric_func_factory(s3_put_bytes_rate_limit_sleep_ns, + s3_put_bytes_rate_limit_sleep_count); + default: + return [](int64_t) {}; + } +} + +// min(per_core * cores, cap) with overflow protection; cap == 0 means no cap. +int64_t cap_multiply(int64_t per_core, int64_t cores, int64_t cap) { + cap = cap > 0 ? cap : std::numeric_limits<int64_t>::max(); + if (per_core > cap / cores) { + return cap; + } + return per_core * cores; +} + +size_t index_of(S3RateLimitType type) { + CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type); + return static_cast<size_t>(type); +} + +} // namespace + +S3EffectiveRateLimit resolve_s3_rate_limit(S3RateLimitType type, int64_t cores) { + const bool is_get = type == S3RateLimitType::GET; + const int64_t qps_per_core = is_get ? config::s3_get_qps_per_core : config::s3_put_qps_per_core; + const int64_t qps_max = is_get ? config::s3_get_qps_max : config::s3_put_qps_max; + const int64_t bytes_per_core = is_get ? config::s3_get_bytes_per_second_per_core + : config::s3_put_bytes_per_second_per_core; + const int64_t bytes_max = + is_get ? config::s3_get_bytes_per_second_max : config::s3_put_bytes_per_second_max; + cores = std::max<int64_t>(1, cores); + + S3EffectiveRateLimit limit; + if (qps_per_core < 0) { + // Unset: the legacy absolute configs stay in charge, bit-for-bit compatible. + limit.qps = is_get ? config::s3_get_token_per_second : config::s3_put_token_per_second; + limit.burst = is_get ? config::s3_get_bucket_tokens : config::s3_put_bucket_tokens; + limit.count_limit = is_get ? config::s3_get_token_limit : config::s3_put_token_limit; + } else if (qps_per_core > 0) { + limit.qps = cap_multiply(qps_per_core, cores, qps_max); + limit.burst = limit.qps; // burst = 1 second worth of quota + } // qps_per_core == 0: QPS limiting disabled, all fields stay 0. + + if (bytes_per_core > 0) { + limit.bytes_per_second = cap_multiply(bytes_per_core, cores, bytes_max); + } + return limit; +} + +int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit) { + if (type == S3RateLimitType::UNKNOWN) { + return -1; + } + return S3RateLimiterManager::instance().qps_limiter(type)->reset(max_speed, max_burst, limit); +} + +int64_t s3_rate_limiter_cpu_cores() { + if (int64_t overridden = config::s3_rate_limiter_cpu_cores; overridden > 0) { + return overridden; + } + int physical = static_cast<int>(std::thread::hardware_concurrency()); + // Re-read the cgroup quota on every call: serverless BEs can be resized in place, + // and the daemon refresh thread picks the change up through here. + int limited = CGroupUtil::get_cgroup_limited_cpu_number(physical); + return std::max(1, limited); +} + +S3RateLimiterManager::S3RateLimiterManager() { + const int64_t cores = s3_rate_limiter_cpu_cores(); + for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) { + auto limit = resolve_s3_rate_limit(type, cores); + _qps_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>( + limit.qps, limit.burst, limit.count_limit, s3_rate_limiter_metric_func(type)); + _bytes_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>( + limit.bytes_per_second, limit.bytes_per_second, 0, + bytes_rate_limiter_metric_func(type)); + } +} + +S3RateLimiterManager& S3RateLimiterManager::instance() { + static S3RateLimiterManager ret; + return ret; +} + +S3RateLimiterHolder* S3RateLimiterManager::qps_limiter(S3RateLimitType type) { + return _qps_limiters[index_of(type)].get(); +} + +S3RateLimiterHolder* S3RateLimiterManager::bytes_limiter(S3RateLimitType type) { + return _bytes_limiters[index_of(type)].get(); +} + +void S3RateLimiterManager::refresh() { + std::lock_guard guard(_refresh_lock); + const int64_t cores = s3_rate_limiter_cpu_cores(); + for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) { + const auto limit = resolve_s3_rate_limit(type, cores); + + auto* qps = qps_limiter(type); + if (qps->get_max_speed() != static_cast<size_t>(limit.qps) || + qps->get_max_burst() != static_cast<size_t>(limit.burst) || + qps->get_limit() != static_cast<size_t>(limit.count_limit)) { + qps->reset(limit.qps, limit.burst, limit.count_limit); + LOG(INFO) << "reset S3 " << to_string(type) << " QPS rate limiter, qps=" << limit.qps + << ", burst=" << limit.burst << ", count_limit=" << limit.count_limit + << ", cores=" << cores; + } + + auto* bytes = bytes_limiter(type); + if (bytes->get_max_speed() != static_cast<size_t>(limit.bytes_per_second)) { + bytes->reset(limit.bytes_per_second, limit.bytes_per_second, 0); + LOG(INFO) << "reset S3 " << to_string(type) + << " bytes rate limiter, bytes_per_second=" << limit.bytes_per_second + << ", cores=" << cores; + } + } +} + +S3RateLimitGuard::S3RateLimitGuard(S3RateLimitType type, size_t estimated_bytes) { + if (!config::enable_s3_rate_limiter) { + return; + } + auto& mgr = S3RateLimiterManager::instance(); + + auto* qps = mgr.qps_limiter(type); + if (qps->is_enabled() && + apply_s3_rate_limit(type, qps, config::s3_rate_limiter_log_interval) < 0) { + _ok = false; + return; + } + + if (estimated_bytes == 0) { + return; + } + auto* bytes = mgr.bytes_limiter(type); + if (!bytes->is_enabled()) { + return; + } + // Clamp the reservation to 1 second worth of bandwidth so a single oversized IO + // (e.g. a whole-file read_at) cannot create unbounded upfront debt. The clamped + // remainder is intentionally not accounted; effective quotas below the single-IO + // upper bound are excluded by the config contract (see config.cpp). + _reserved = std::min(estimated_bytes, bytes->get_max_speed()); + if (_reserved > 0) { + // Debt model: may sleep, never rejects (count limit is 0). Pin the charged + // bucket generation for settle(). + _charged_bucket = bytes->charge(_reserved); Review Comment: [P2] Size and charge against the same bucket generation `get_max_speed()` and `charge()` snapshot the holder separately, so refresh can swap generations between them. If a 50 MiB request reads an old 1 MiB/s rate and then charges a fresh 100 MiB/s bucket, it accounts only 1 MiB instead of 50 MiB; reversing the rates can charge 50 MiB to a fresh 1 MiB/s bucket and sleep about 49 seconds despite the intended one-second clamp. Pinning the pointer returned by `charge()` does not pin the generation used to compute `_reserved`. Please expose one holder operation that snapshots a generation, computes the clamp, charges it, and returns that same pointer/reserved amount, with barrier tests for both rate directions. ########## be/src/util/s3_rate_limiter_manager.cpp: ########## @@ -0,0 +1,204 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "util/s3_rate_limiter_manager.h" + +#include <algorithm> +#include <limits> +#include <thread> + +#include "common/config.h" +#include "common/logging.h" +#include "util/cgroup_util.h" + +namespace doris { + +bvar::Adder<int64_t> s3_get_bytes_rate_limit_sleep_ns("s3_get_bytes_rate_limit_sleep_ns"); +bvar::Adder<int64_t> s3_get_bytes_rate_limit_sleep_count("s3_get_bytes_rate_limit_sleep_count"); +bvar::Adder<int64_t> s3_put_bytes_rate_limit_sleep_ns("s3_put_bytes_rate_limit_sleep_ns"); +bvar::Adder<int64_t> s3_put_bytes_rate_limit_sleep_count("s3_put_bytes_rate_limit_sleep_count"); + +namespace { + +std::function<void(int64_t)> bytes_rate_limiter_metric_func(S3RateLimitType type) { + switch (type) { + case S3RateLimitType::GET: + return metric_func_factory(s3_get_bytes_rate_limit_sleep_ns, + s3_get_bytes_rate_limit_sleep_count); + case S3RateLimitType::PUT: + return metric_func_factory(s3_put_bytes_rate_limit_sleep_ns, + s3_put_bytes_rate_limit_sleep_count); + default: + return [](int64_t) {}; + } +} + +// min(per_core * cores, cap) with overflow protection; cap == 0 means no cap. +int64_t cap_multiply(int64_t per_core, int64_t cores, int64_t cap) { + cap = cap > 0 ? cap : std::numeric_limits<int64_t>::max(); + if (per_core > cap / cores) { + return cap; + } + return per_core * cores; +} + +size_t index_of(S3RateLimitType type) { + CHECK(type == S3RateLimitType::GET || type == S3RateLimitType::PUT) << to_string(type); + return static_cast<size_t>(type); +} + +} // namespace + +S3EffectiveRateLimit resolve_s3_rate_limit(S3RateLimitType type, int64_t cores) { + const bool is_get = type == S3RateLimitType::GET; + const int64_t qps_per_core = is_get ? config::s3_get_qps_per_core : config::s3_put_qps_per_core; + const int64_t qps_max = is_get ? config::s3_get_qps_max : config::s3_put_qps_max; + const int64_t bytes_per_core = is_get ? config::s3_get_bytes_per_second_per_core + : config::s3_put_bytes_per_second_per_core; + const int64_t bytes_max = + is_get ? config::s3_get_bytes_per_second_max : config::s3_put_bytes_per_second_max; + cores = std::max<int64_t>(1, cores); + + S3EffectiveRateLimit limit; + if (qps_per_core < 0) { + // Unset: the legacy absolute configs stay in charge, bit-for-bit compatible. + limit.qps = is_get ? config::s3_get_token_per_second : config::s3_put_token_per_second; + limit.burst = is_get ? config::s3_get_bucket_tokens : config::s3_put_bucket_tokens; + limit.count_limit = is_get ? config::s3_get_token_limit : config::s3_put_token_limit; + } else if (qps_per_core > 0) { + limit.qps = cap_multiply(qps_per_core, cores, qps_max); + limit.burst = limit.qps; // burst = 1 second worth of quota + } // qps_per_core == 0: QPS limiting disabled, all fields stay 0. + + if (bytes_per_core > 0) { + limit.bytes_per_second = cap_multiply(bytes_per_core, cores, bytes_max); + } + return limit; +} + +int reset_s3_rate_limiter(S3RateLimitType type, size_t max_speed, size_t max_burst, size_t limit) { + if (type == S3RateLimitType::UNKNOWN) { + return -1; + } + return S3RateLimiterManager::instance().qps_limiter(type)->reset(max_speed, max_burst, limit); +} + +int64_t s3_rate_limiter_cpu_cores() { + if (int64_t overridden = config::s3_rate_limiter_cpu_cores; overridden > 0) { + return overridden; + } + int physical = static_cast<int>(std::thread::hardware_concurrency()); + // Re-read the cgroup quota on every call: serverless BEs can be resized in place, + // and the daemon refresh thread picks the change up through here. + int limited = CGroupUtil::get_cgroup_limited_cpu_number(physical); + return std::max(1, limited); +} + +S3RateLimiterManager::S3RateLimiterManager() { + const int64_t cores = s3_rate_limiter_cpu_cores(); + for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) { + auto limit = resolve_s3_rate_limit(type, cores); + _qps_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>( + limit.qps, limit.burst, limit.count_limit, s3_rate_limiter_metric_func(type)); + _bytes_limiters[index_of(type)] = std::make_unique<S3RateLimiterHolder>( + limit.bytes_per_second, limit.bytes_per_second, 0, + bytes_rate_limiter_metric_func(type)); + } +} + +S3RateLimiterManager& S3RateLimiterManager::instance() { + static S3RateLimiterManager ret; + return ret; +} + +S3RateLimiterHolder* S3RateLimiterManager::qps_limiter(S3RateLimitType type) { + return _qps_limiters[index_of(type)].get(); +} + +S3RateLimiterHolder* S3RateLimiterManager::bytes_limiter(S3RateLimitType type) { + return _bytes_limiters[index_of(type)].get(); +} + +void S3RateLimiterManager::refresh() { + std::lock_guard guard(_refresh_lock); + const int64_t cores = s3_rate_limiter_cpu_cores(); + for (auto type : {S3RateLimitType::GET, S3RateLimitType::PUT}) { + const auto limit = resolve_s3_rate_limit(type, cores); + + auto* qps = qps_limiter(type); + if (qps->get_max_speed() != static_cast<size_t>(limit.qps) || + qps->get_max_burst() != static_cast<size_t>(limit.burst) || + qps->get_limit() != static_cast<size_t>(limit.count_limit)) { + qps->reset(limit.qps, limit.burst, limit.count_limit); + LOG(INFO) << "reset S3 " << to_string(type) << " QPS rate limiter, qps=" << limit.qps + << ", burst=" << limit.burst << ", count_limit=" << limit.count_limit + << ", cores=" << cores; + } + + auto* bytes = bytes_limiter(type); + if (bytes->get_max_speed() != static_cast<size_t>(limit.bytes_per_second)) { + bytes->reset(limit.bytes_per_second, limit.bytes_per_second, 0); + LOG(INFO) << "reset S3 " << to_string(type) + << " bytes rate limiter, bytes_per_second=" << limit.bytes_per_second + << ", cores=" << cores; + } + } +} + +S3RateLimitGuard::S3RateLimitGuard(S3RateLimitType type, size_t estimated_bytes) { + if (!config::enable_s3_rate_limiter) { + return; + } + auto& mgr = S3RateLimiterManager::instance(); + + auto* qps = mgr.qps_limiter(type); + if (qps->is_enabled() && + apply_s3_rate_limit(type, qps, config::s3_rate_limiter_log_interval) < 0) { + _ok = false; + return; + } + + if (estimated_bytes == 0) { + return; + } + auto* bytes = mgr.bytes_limiter(type); + if (!bytes->is_enabled()) { + return; + } + // Clamp the reservation to 1 second worth of bandwidth so a single oversized IO + // (e.g. a whole-file read_at) cannot create unbounded upfront debt. The clamped + // remainder is intentionally not accounted; effective quotas below the single-IO + // upper bound are excluded by the config contract (see config.cpp). + _reserved = std::min(estimated_bytes, bytes->get_max_speed()); Review Comment: [P1] Account bytes beyond the one-second reservation The clamp permanently drops the remainder instead of merely bounding the wait. For example, at 1 MiB/s a 16 MiB `get_object()` charges 1 MiB, transfers all 16 MiB, and `settle(16 MiB)` does nothing because the actual size exceeds `_reserved`. This is reachable through whole-file `download_impl()` (and writer/read buffers can likewise exceed the rate), while the validators accept rates down to one byte and cannot enforce the comment's assumed I/O bound. Thus one successful call can exceed the configured bandwidth arbitrarily; SDK retries only multiply this separate issue. Please account the full estimate/actual size in bounded chunks or charge the remainder, and cover a production-path I/O larger than the effective rate. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
