github-actions[bot] commented on code in PR #65694: URL: https://github.com/apache/doris/pull/65694#discussion_r3636083928
########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java: ########## @@ -0,0 +1,278 @@ +// 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. + +package org.apache.doris.cloud.rpc; + +import org.apache.doris.common.Config; +import org.apache.doris.common.ConfigException; +import org.apache.doris.common.MetaServiceRpcRateLimitConfigValidator; +import org.apache.doris.metric.CloudMetrics; +import org.apache.doris.metric.MetricRepo; +import org.apache.doris.rpc.RpcException; + +import com.google.common.collect.Maps; +import io.github.resilience4j.ratelimiter.RateLimiter; +import io.github.resilience4j.ratelimiter.RateLimiterConfig; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.time.Duration; +import java.util.Collections; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; + +class MetaServiceRpcRateLimiter { + private static final Logger LOG = LogManager.getLogger(MetaServiceRpcRateLimiter.class); + + private static final int CPU_CORES = Runtime.getRuntime().availableProcessors(); + private static final RateLimitConfigSnapshot EMPTY_SNAPSHOT = + new RateLimitConfigSnapshot(0, "", 1, 0); + + private final ReentrantLock configLock = new ReentrantLock(); + private final ConcurrentMap<String, RateLimiterHolder> rateLimiters = Maps.newConcurrentMap(); + private volatile RateLimitConfigSnapshot currentSnapshot = EMPTY_SNAPSHOT; + private volatile Map<String, Integer> methodQpsPerCore = Collections.emptyMap(); + + long acquire(String methodName) throws RpcException { + return acquire(methodName, 1); + } + + long acquire(String methodName, int permits) throws RpcException { + if (!Config.meta_service_rpc_rate_limit_enabled) { + return 0; + } + + RateLimiterHolder holder = getRateLimiter(methodName); + if (holder == null) { + return 0; + } + + int permitsToAcquire = Math.min(Math.max(permits, 1), holder.maxPermitsInTimeout); + // Resilience4j returns negative when the estimated wait exceeds the configured timeout. + // Otherwise the returned wait time is within meta_service_rpc_rate_limit_wait_timeout_ms. + long nanosToWait = holder.rateLimiter.reservePermission(permitsToAcquire); + if (nanosToWait < 0) { + throw new MetaServiceRateLimitException(methodName, + Config.meta_service_rpc_rate_limit_wait_timeout_ms); + } + if (nanosToWait == 0) { + return 0; + } + + long waitMs = TimeUnit.NANOSECONDS.toMillis(nanosToWait); + if (LOG.isDebugEnabled()) { + LOG.debug("meta service rpc rate limiter waits before acquiring permission, method: {}, permits: {}, " + + "original permits: {}, max permits in timeout: {}, limit for period: {}, " + + "burst seconds: {}, wait ms: {}", + methodName, permitsToAcquire, permits, holder.maxPermitsInTimeout, holder.limitForPeriod, + holder.burstSeconds, + waitMs); + } + long waitStartNs = System.nanoTime(); + try { + TimeUnit.NANOSECONDS.sleep(nanosToWait); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RpcException("", e.getMessage(), e); + } + long actualWaitNs = System.nanoTime() - waitStartNs; + if (MetricRepo.isInit && Config.isCloudMode()) { + CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY.getOrAdd(methodName) + .update(TimeUnit.NANOSECONDS.toMillis(actualWaitNs)); + } + return actualWaitNs; + } + + private RateLimiterHolder getRateLimiter(String methodName) throws RpcException { + refreshConfigIfNeeded(); + RateLimitConfigSnapshot snapshot = currentSnapshot; + int qpsPerCore = methodQpsPerCore.getOrDefault(methodName, snapshot.defaultQpsPerCore); + if (qpsPerCore <= 0) { + rateLimiters.remove(methodName); + return null; + } + + int limitForPeriod = getLimitForPeriod(methodName, qpsPerCore, snapshot.burstSeconds); + int maxPermitsInTimeout = getMaxPermitsInTimeout(methodName, limitForPeriod, snapshot.burstSeconds, + snapshot.waitTimeoutMs); + RateLimiterHolder holder = rateLimiters.compute(methodName, (name, existingHolder) -> { Review Comment: [P2] Keep matching holders off the map-update path `getRateLimiter()` reaches `rateLimiters.compute()` on every enabled RPC even when the snapshot is unchanged and the existing holder already matches. Because `ConcurrentHashMap.compute` performs the remapping atomically for this method key, concurrent unit-weight calls such as `getPartitionVersion` all funnel through the same map update before reaching Resilience4j's CAS-based limiter; overload/rejection traffic is serialized there as well. Please add a generation-safe `get()` fast path for a matching holder and reserve the fenced `compute` path for installation/reconfiguration, with a concurrent steady-state test or benchmark. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java: ########## @@ -207,6 +256,7 @@ public <Response> Response executeRequest(String methodName, Function<MetaServic MetaServiceClient client = null; boolean requestFailed = false; try { + acquireRateLimit(methodName, 1); Review Comment: [P1] Charge each transport retry to the limiter This acquisition only covers the single wrapper invocation, but the production OSS channel enables `retrying_service_config.json` for all `doris.cloud.MetaService` methods; an `UNAVAILABLE` result can therefore make the blocking stub perform up to 10 physical attempts below `function.apply()` without returning here. The async get-version stub has the same gap. During an outage/recovery, one permit can thus fan out into ten requests and retry cohorts can exceed the configured QPS precisely when MetaService is unhealthy. Please make every physical retry reacquire a permit (for example, disable channel-level retries and use the bounded wrapper retry, or add an attempt-aware interceptor) and cover both blocking and async calls with an in-process gRPC retry test. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java: ########## @@ -313,6 +361,7 @@ public Future<Cloud.GetVersionResponse> getVisibleVersionAsync(Cloud.GetVersionR } try { + recordGetVersionRateLimitWait(acquireRateLimit(methodName, getGetVersionRateLimitPermits(request))); Review Comment: [P1] Do not launch get-version after its caller deadline `VersionHelper.getVisibleVersionInternal()` starts its 3-second attempt deadline before entering this method, but this acquisition may legally sleep for the independently configurable limiter timeout and still dispatch the RPC afterward. For example, after the current period's permits are exhausted, with 1 QPS/core, a 1-second burst, and a 5-second limiter timeout, a `4 * CPU_CORES` batch can reserve across roughly four cycles; once this line returns, the helper calls `future.get(0, ...)`, drops the still-live 30-second RPC on `TimeoutException`, and starts another outer retry. That creates overlapping requests during saturation. Please share one end-to-end deadline with acquisition and dispatch (or explicitly start the network deadline after acquisition), avoid dispatch once the chosen deadline is exhausted, retire or cancel timed-out futures without treating caller cancellation as a shared-client failure, and add a caller-level multi-cycle-wait test. -- 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]
