[feature](fe) Add meta service RPC rate limiting in FE

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3477,6 +3477,38 @@ public static int metaServiceRpcRetryTimes() {
  "In cloud mode, the retry count when the FE request to meta service times out. Default is 1."})
  public static int meta_service_rpc_timeout_retry_times = 1;
 
+ @ConfField(mutable = true, description = {
+ "Whether to enable QPS rate limit for RPC requests to meta service."})
+ public static boolean meta_service_rpc_rate_limit_enabled = false;
+
+ @ConfField(mutable = true, description = {
+ "Default QPS limit for each method (requests per second) in each cpu core, "
+ + "non-positive value (<= 0) means no limit"})
+ public static int meta_service_rpc_rate_limit_default_qps_per_core = 50;
+
+ @ConfField(mutable = true,
+ callback = MetaServiceRpcRateLimitConfigValidator.QpsConfigHandler.class,
+ description = {
+ "QPS limit config per rpc method to meta service in per cpu core, "
+ + "format: method1:qps1;method2:qps2, "
+ + "e.g.: getPartitionVersion:100;getTableVersion:100;getTabletStats:50, "
+ + "non-positive value (<= 0) means no limit"})
+ public static String meta_service_rpc_rate_limit_qps_per_core_config
+ = "getPartitionVersion:500;getTableVersion:500;getTabletStats:50;beginTxn:50";
+
+ @ConfField(mutable = true,
+ callback = MetaServiceRpcRateLimitConfigValidator.PositiveIntConfigHandler.class,
+ description = {
+ "Burst window for meta service RPC rate limit in seconds. "
+ + "The long-term average QPS is unchanged, while calls can burst within this window."})
+ public static int meta_service_rpc_rate_limit_burst_seconds = 2;
+
+ @ConfField(mutable = true, callback = MetaServiceRpcRateLimitConfigValidator.NonNegativeLongConfigHandler.class,
+ description = {
+ "Max wait time in milliseconds when meta service RPC is rate limited, "
+ + "zero means fail fast."})
+ public static long meta_service_rpc_rate_limit_wait_timeout_ms = 1000;
+
  @ConfField(mutable = true, description = {
  "In cloud mode, the auto start and stop ignores the databases used by internal jobs, "
  + "such as those used for statistics. "
@@ -3723,4 +3755,5 @@ public static int metaServiceRpcRetryTimes() {
  + "（持有主副本的桶），并在单个 tablet 写入量超过阈值（默认 200 MB）后在本地桶之间轮转。"
  + "可降低导入内存压力并提升随机分桶表的吞吐量，覆盖所有导入类型。"})
  public static boolean enable_adaptive_random_bucket_load = true;
+
 }

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));
+ super.handle(field, trimmedVal);
+ }
+ }
+}

fe/fe-core/pom.xml

@@ -228,6 +228,10 @@ under the License.
  <artifactId>guava-testlib</artifactId>
  <scope>test</scope>
  </dependency>
+ <dependency>
+ <groupId>io.github.resilience4j</groupId>
+ <artifactId>resilience4j-ratelimiter</artifactId>
+ </dependency>
  <!-- https://mvnrepository.com/artifact/com.googlecode.java-ipv6/java-ipv6 -->
  <dependency>
  <groupId>com.googlecode.java-ipv6</groupId>

fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.cloud.proto.Cloud;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.profile.SummaryProfile;
 import org.apache.doris.metric.CloudMetrics;
 import org.apache.doris.metric.MetricRepo;
 import org.apache.doris.rpc.RpcException;
@@ -39,6 +40,7 @@
 
 public class MetaServiceProxy {
  private static final Logger LOG = LogManager.getLogger(MetaServiceProxy.class);
+ private static final MetaServiceRpcRateLimiter META_SERVICE_RPC_RATE_LIMITER = new MetaServiceRpcRateLimiter();
 
  // use exclusive lock to make sure only one thread can add or remove client from
  // serviceMap.
@@ -105,24 +107,71 @@ public Cloud.GetInstanceResponse getInstance(Cloud.GetInstanceRequest request)
  }
 
  try {
+ acquireRateLimit(methodName);
  final MetaServiceClient client = getProxy();
  Cloud.GetInstanceResponse response = client.getInstance(request);
  if (MetricRepo.isInit && Config.isCloudMode()) {
  CloudMetrics.META_SERVICE_RPC_LATENCY.getOrAdd(methodName)
  .update(System.currentTimeMillis() - startTime);
  }
  return response;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
  } catch (Exception e) {
- 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);
  throw new RpcException("", e.getMessage(), e);
  }
  }
 
+ private static long acquireRateLimit(String methodName) throws RpcException {
+ return META_SERVICE_RPC_RATE_LIMITER.acquire(methodName);
+ }
+
+ private static long acquireRateLimit(String methodName, int permits) throws RpcException {
+ return META_SERVICE_RPC_RATE_LIMITER.acquire(methodName, permits);
+ }
+
+ private static int getGetVersionRateLimitPermits(Cloud.GetVersionRequest request) {
+ if (!request.getBatchMode()) {
+ return 1;
+ }
+ int permits = request.hasIsTableVersion() && request.getIsTableVersion()
+ ? request.getTableIdsCount()
+ : request.getPartitionIdsCount();
+ return Math.max(permits, 1);
+ }
+
+ static void resetMetaServiceRpcRateLimitForTest() {
+ META_SERVICE_RPC_RATE_LIMITER.reset();
+ }
+
+ private static void recordRpcFailed(String methodName, long startTime) {
+ 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);
+ }
+ }
+
+ private static void recordRpcRateLimited(String methodName) {
+ if (MetricRepo.isInit && Config.isCloudMode()) {
+ CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED.increase(1L);
+ CloudMetrics.META_SERVICE_RPC_RATE_LIMITED.getOrAdd(methodName).increase(1L);
+ }
+ }
+
+ private static void recordGetVersionRateLimitWait(long waitNs) {
+ if (waitNs <= 0) {
+ return;
+ }
+ SummaryProfile profile = SummaryProfile.getSummaryProfile(null);
+ if (profile != null) {
+ profile.addGetMetaVersionRateLimitWaitTime(waitNs);
+ }
+ }
+
  public void removeProxy(String address) {
  LOG.warn("begin to remove proxy: {}", address);
  MetaServiceClient service;
@@ -207,6 +256,7 @@ public <Response> Response executeRequest(String methodName, Function<MetaServic
  MetaServiceClient client = null;
  boolean requestFailed = false;
  try {
+ acquireRateLimit(methodName, 1);
  client = proxy.getProxy();
  if (tried > 1 && MetricRepo.isInit && Config.isCloudMode()) {
  CloudMetrics.META_SERVICE_RPC_ALL_RETRY.increase(1L);
@@ -289,13 +339,11 @@ private <Response> Response executeWithMetrics(String methodName, Function<MetaS
  .update(System.currentTimeMillis() - startTime);
  }
  return response;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
  } catch (RpcException e) {
- 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);
  throw e;
  }
  }
@@ -313,6 +361,7 @@ public Future<Cloud.GetVersionResponse> getVisibleVersionAsync(Cloud.GetVersionR
  }
 
  try {
+ recordGetVersionRateLimitWait(acquireRateLimit(methodName, getGetVersionRateLimitPermits(request)));
  client = getProxy();
  Future<Cloud.GetVersionResponse> future = client.getVisibleVersionAsync(request);
  if (future instanceof com.google.common.util.concurrent.ListenableFuture) {
@@ -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);
  }
  }
  }, com.google.common.util.concurrent.MoreExecutors.directExecutor());
  }
  return future;
+ } catch (MetaServiceRateLimitException e) {
+ recordRpcRateLimited(methodName);
+ throw e;
  } catch (Exception e) {
- 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 (client != null) {
  client.shutdown(true);
  }
  throw new RpcException("", e.getMessage(), e);
  }
  }
 
- public Cloud.GetVersionResponse getVersion(Cloud.GetVersionRequest request) throws RpcException {
- String methodName = request.hasIsTableVersion() && request.getIsTableVersion() ? "getTableVersion"
- : "getPartitionVersion";
- return executeWithMetrics(methodName, (client) -> client.getVersion(request),
- Cloud.GetVersionResponse::getStatus);
- }
-
  public Cloud.CreateTabletsResponse createTablets(Cloud.CreateTabletsRequest request) throws RpcException {
  return executeWithMetrics("createTablets", (client) -> client.createTablets(request),
  Cloud.CreateTabletsResponse::getStatus);

fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRateLimitException.java

@@ -0,0 +1,27 @@
+// 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.rpc.RpcException;
+
+class MetaServiceRateLimitException extends RpcException {
+ MetaServiceRateLimitException(String methodName, long waitTimeoutMs) {
+ super("", "meta service rpc rate limited, method: " + methodName
+ + ", wait timeout: " + waitTimeoutMs + "ms");
+ }
+}