github-actions[bot] commented on code in PR #65694: URL: https://github.com/apache/doris/pull/65694#discussion_r3593894413
########## fe/fe-common/src/main/java/org/apache/doris/common/MetaServiceRpcRateLimitConfigValidator.java: ########## @@ -0,0 +1,103 @@ +// 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.common; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +public final class MetaServiceRpcRateLimitConfigValidator { + private MetaServiceRpcRateLimitConfigValidator() { + } + + public static Map<String, Integer> parseQpsPerCoreConfig(String config) throws ConfigException { + if (config == null || config.trim().isEmpty()) { + return Collections.emptyMap(); + } + + Map<String, Integer> qpsPerCore = new HashMap<>(); + Set<String> methods = new HashSet<>(); + for (String item : config.split(";")) { + String trimmedItem = item.trim(); + if (trimmedItem.isEmpty()) { + continue; + } + int separatorIndex = trimmedItem.indexOf(':'); + if (separatorIndex <= 0 || separatorIndex != trimmedItem.lastIndexOf(':') + || separatorIndex == trimmedItem.length() - 1) { + throw new ConfigException("Invalid format, expected method1:qps1;method2:qps2"); + } + + String methodName = trimmedItem.substring(0, separatorIndex).trim(); + String qpsText = trimmedItem.substring(separatorIndex + 1).trim(); + if (methodName.isEmpty() || qpsText.isEmpty()) { + throw new ConfigException("Invalid format, expected method1:qps1;method2:qps2"); + } + if (!methods.add(methodName)) { + throw new ConfigException("Duplicate method: " + methodName); + } + try { + qpsPerCore.put(methodName, Integer.parseInt(qpsText)); + } catch (NumberFormatException e) { + throw new ConfigException("Invalid qps for method: " + methodName, e); + } + } + return qpsPerCore; + } + + public static void validatePositive(String fieldName, int value) throws ConfigException { + if (value <= 0) { + throw new ConfigException(fieldName + " must be positive"); + } + } + + public static void validateNonNegative(String fieldName, long value) throws ConfigException { + if (value < 0) { + throw new ConfigException(fieldName + " must be non-negative"); + } + } + + public static class QpsConfigHandler extends ConfigBase.DefaultConfHandler { + @Override + public void handle(Field field, String confVal) throws Exception { + parseQpsPerCoreConfig(confVal); + super.handle(field, confVal); + } + } + + public static class PositiveIntConfigHandler extends ConfigBase.DefaultConfHandler { + @Override + public void handle(Field field, String confVal) throws Exception { + String trimmedVal = confVal.trim(); + validatePositive(field.getName(), Integer.parseInt(trimmedVal)); + super.handle(field, trimmedVal); + } + } + + public static class NonNegativeLongConfigHandler extends ConfigBase.DefaultConfHandler { + @Override + public void handle(Field field, String confVal) throws Exception { + String trimmedVal = confVal.trim(); + validateNonNegative(field.getName(), Long.parseLong(trimmedVal)); Review Comment: [P2] Validate the timeout against the dependency's duration bound This accepts every nonnegative `long`, but Resilience4j 2.4.0 immediately converts the timeout `Duration` to nanoseconds when building the limiter. For example, `9223372036855` ms is accepted and published here, then the next limited RPC fails locally with `RuntimeException: TimeoutDuration too large` before contacting meta service. Reject values above `Long.MAX_VALUE / 1_000_000` in both mutable and runtime/startup validation (or convert the library failure before publication), with boundary tests. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java: ########## @@ -0,0 +1,253 @@ +// 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.limitForPeriod); + // 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: {}, " + + "limit for period: {}, burst seconds: {}, wait ms: {}", + methodName, permitsToAcquire, 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); + RateLimiterHolder holder = rateLimiters.compute(methodName, (name, existingHolder) -> { + if (existingHolder != null && existingHolder.matches(limitForPeriod, snapshot.burstSeconds, + snapshot.waitTimeoutMs)) { + return existingHolder; + } + return new RateLimiterHolder(createRateLimiter(methodName, limitForPeriod, + snapshot.burstSeconds, snapshot.waitTimeoutMs), limitForPeriod, + snapshot.burstSeconds, snapshot.waitTimeoutMs); + }); + return holder; + } + + private void refreshConfigIfNeeded() throws RpcException { + RateLimitConfigSnapshot latestSnapshot = RateLimitConfigSnapshot.current(); + if (latestSnapshot.equals(currentSnapshot)) { + return; + } + + configLock.lock(); + try { + latestSnapshot = RateLimitConfigSnapshot.current(); + if (latestSnapshot.equals(currentSnapshot)) { + return; + } + validateConfig(latestSnapshot); + methodQpsPerCore = parseMethodQpsPerCore(latestSnapshot.qpsPerCoreConfig); + currentSnapshot = latestSnapshot; + } finally { + configLock.unlock(); + } + } + + private void validateConfig(RateLimitConfigSnapshot snapshot) throws RpcException { + try { + MetaServiceRpcRateLimitConfigValidator.validatePositive("meta_service_rpc_rate_limit_burst_seconds", + snapshot.burstSeconds); + MetaServiceRpcRateLimitConfigValidator.validateNonNegative("meta_service_rpc_rate_limit_wait_timeout_ms", + snapshot.waitTimeoutMs); + } catch (ConfigException e) { + throw new RpcException("", e.getMessage(), e); + } + } + + private Map<String, Integer> parseMethodQpsPerCore(String config) throws RpcException { + try { + return MetaServiceRpcRateLimitConfigValidator.parseQpsPerCoreConfig(config); + } catch (ConfigException e) { + throw new RpcException("", "invalid meta_service_rpc_rate_limit_qps_per_core_config: " + config, e); + } + } + + private int getLimitForPeriod(String methodName, int qpsPerCore, int burstSeconds) throws RpcException { + long limitForPeriod = (long) qpsPerCore * CPU_CORES * burstSeconds; Review Comment: [P2] Check the product before the `long` multiplication can overflow The guard runs after all three factors have already been multiplied. With `qpsPerCore = burstSeconds = Integer.MAX_VALUE` on a 4-core FE, the `long` wraps to `-17179869180`, the `> Integer.MAX_VALUE` check is bypassed, and the cast yields a valid `limitForPeriod` of 4. The config update thus succeeds but effectively permits four calls per roughly 68-year refresh period. Use checked multiplication or division-based preconditions before computing the product, and add a max-value regression case. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java: ########## @@ -0,0 +1,253 @@ +// 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.limitForPeriod); Review Comment: [P1] Preserve the full weight of oversized version batches This clamp makes every batch larger than one period cost exactly `limitForPeriod` permits. Production already exceeds that boundary: `CloudSyncVersionDaemon` sends up to 2,000 entries by default, while the default capacity is `100 * CPU_CORES * 2` (800 on a 4-core FE), and query partition batches can be larger still. Consequently a 2,000- or 20,000-item request is charged the same 800 permits, so the configured weighted QPS no longer bounds the most expensive meta-service work. Resilience4j supports multi-cycle permit reservations; please reserve/account the full item count (or explicitly chunk it) and let the configured timeout reject debt that cannot fit. The `permits > limitForPeriod` test should verify proportional debt rather than this truncation. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java: ########## @@ -331,40 +380,26 @@ public void onSuccess(Cloud.GetVersionResponse result) { @Override public void onFailure(Throwable t) { - if (MetricRepo.isInit && Config.isCloudMode()) { - CloudMetrics.META_SERVICE_RPC_ALL_FAILED.increase(1L); - CloudMetrics.META_SERVICE_RPC_FAILED.getOrAdd(methodName).increase(1L); - CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName) - .update(System.currentTimeMillis() - startTime); - } + recordRpcFailed(methodName, startTime); if (finalClient != null) { finalClient.shutdown(true); Review Comment: [P1] Keep rate-limit rejection out of the version retry loop Rethrowing the dedicated exception here is not enough for the production path: `VersionHelper.getVisibleVersionInternal()` catches every `RpcException` and returns `null`, so `getVisibleVersion()` can run the default 200-attempt retry loop. Under sustained saturation with `meta_service_rpc_rate_limit_wait_timeout_ms=0`, every acquisition is rejected immediately, yet the logical call still sleeps after each attempt, totaling about 52 to under 120 seconds while query planners and sync workers generate 200 local attempts/logs each. Please propagate `MetaServiceRateLimitException` through `VersionHelper` (or give it an explicitly bounded limiter-specific policy) and add an end-to-end helper test that observes one prompt rejection. ########## fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java: ########## @@ -0,0 +1,253 @@ +// 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.limitForPeriod); + // 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: {}, " + + "limit for period: {}, burst seconds: {}, wait ms: {}", + methodName, permitsToAcquire, 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); + RateLimiterHolder holder = rateLimiters.compute(methodName, (name, existingHolder) -> { Review Comment: [P2] Guard every holder mutation against stale snapshots A request can capture an old snapshot before a dynamic update and mutate `rateLimiters` after a newer request has installed and consumed the new holder. A stale positive snapshot can replace it in `compute`; a stale unlimited snapshot can remove it and make its own RPC unthrottled. The next post-update request then installs another fresh holder, resetting the new quota, and multiple pre-update callers can repeat the cycle. Make both removal and installation conditional on the snapshot/generation still being current (retry on mismatch), and add a latch/barrier test that includes an unlimited-to-limited transition. -- 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]
