This is an automated email from the ASF dual-hosted git repository.
mymeiyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c82920d5481 [feature](fe) Add meta service RPC rate limit dry-run mode
(#66940)
c82920d5481 is described below
commit c82920d5481bb66140eeb7f75a65259f99b41161
Author: meiyi <[email protected]>
AuthorDate: Tue Aug 25 17:51:28 2026 +0800
[feature](fe) Add meta service RPC rate limit dry-run mode (#66940)
Enabling FE-side meta service RPC rate limiting immediately waits for or
rejects requests, so operators cannot evaluate whether the configured
limits fit production traffic first.
Add a dry-run mode that evaluates the shared rate limiter and reports
would-wait and would-reject decisions without delaying or rejecting
RPCs.
| Metric | Meaning in dry-run mode |
|---|---|
| `doris_fe_meta_service_rpc_all_rate_limited` | Estimated total number
of rejected requests across all methods |
| `doris_fe_meta_service_rpc_rate_limited{method="..."}` | Estimated
number of rejected requests for each method |
| `doris_fe_meta_service_rpc_rate_limit_wait_latency_ms{method="..."}` |
Estimated wait time and quantiles for each method |
|
`doris_fe_meta_service_rpc_rate_limit_wait_latency_ms_count{method="..."}`
| Estimated number of waits for each method |
| `doris_fe_meta_service_rpc_per_second{method="..."}` | Current QPS for
each method |
---
.../main/java/org/apache/doris/common/Config.java | 7 ++-
.../apache/doris/cloud/rpc/MetaServiceProxy.java | 10 ----
.../doris/cloud/rpc/MetaServiceRpcRateLimiter.java | 30 ++++++++++-
.../doris/cloud/rpc/MetaServiceProxyTest.java | 5 ++
.../cloud/rpc/MetaServiceRpcRateLimiterTest.java | 59 ++++++++++++++++++++++
5 files changed, 99 insertions(+), 12 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index c2998f89f34..b28c4e7398b 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -3422,7 +3422,12 @@ public class Config extends ConfigBase {
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;
+ public static boolean meta_service_rpc_rate_limit_enabled = true;
+
+ @ConfField(mutable = true, description = "Whether to only evaluate and
report meta service RPC rate limits "
+ + "without waiting or rejecting requests. This takes effect only
when meta service RPC rate limiting "
+ + "is enabled.")
+ public static boolean meta_service_rpc_rate_limit_dry_run = true;
@ConfField(mutable = true, description = "Default QPS limit for each
method (requests per second) in each cpu "
+ "core, non-positive value (<= 0) means no limit")
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
index 991bd15bb1f..e23bf6d5ddf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceProxy.java
@@ -116,7 +116,6 @@ public class MetaServiceProxy {
}
return response;
} catch (MetaServiceRateLimitException e) {
- recordRpcRateLimited(methodName);
throw e;
} catch (Exception e) {
recordRpcFailed(methodName, startTime);
@@ -155,13 +154,6 @@ public class MetaServiceProxy {
}
}
- 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;
@@ -340,7 +332,6 @@ public class MetaServiceProxy {
}
return response;
} catch (MetaServiceRateLimitException e) {
- recordRpcRateLimited(methodName);
throw e;
} catch (RpcException e) {
recordRpcFailed(methodName, startTime);
@@ -389,7 +380,6 @@ public class MetaServiceProxy {
}
return future;
} catch (MetaServiceRateLimitException e) {
- recordRpcRateLimited(methodName);
throw e;
} catch (Exception e) {
recordRpcFailed(methodName, startTime);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
index cfcd9407641..7efcfef084b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiter.java
@@ -68,7 +68,22 @@ class MetaServiceRpcRateLimiter {
// 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);
+ boolean dryRun = Config.meta_service_rpc_rate_limit_dry_run;
if (nanosToWait < 0) {
+ if (MetricRepo.isInit) {
+ CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED.increase(1L);
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMITED.getOrAdd(methodName).increase(1L);
+ }
+ if (dryRun) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("meta service rpc rate limiter dry run would
reject request, method: {}, permits: {}, "
+ + "original permits: {}, max permits in
timeout: {}, limit for period: {}, "
+ + "burst seconds: {}, wait timeout ms: {}",
+ methodName, permitsToAcquire, permits,
holder.maxPermitsInTimeout, holder.limitForPeriod,
+ holder.burstSeconds, holder.waitTimeoutMs);
+ }
+ return 0;
+ }
throw new MetaServiceRateLimitException(methodName,
Config.meta_service_rpc_rate_limit_wait_timeout_ms);
}
@@ -77,6 +92,19 @@ class MetaServiceRpcRateLimiter {
}
long waitMs = TimeUnit.NANOSECONDS.toMillis(nanosToWait);
+ if (dryRun) {
+ if (MetricRepo.isInit) {
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY.getOrAdd(methodName).update(waitMs);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("meta service rpc rate limiter dry run would wait
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);
+ }
+ return 0;
+ }
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: {}, "
@@ -93,7 +121,7 @@ class MetaServiceRpcRateLimiter {
throw new RpcException("", e.getMessage(), e);
}
long actualWaitNs = System.nanoTime() - waitStartNs;
- if (MetricRepo.isInit && Config.isCloudMode()) {
+ if (MetricRepo.isInit) {
CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY.getOrAdd(methodName)
.update(TimeUnit.NANOSECONDS.toMillis(actualWaitNs));
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
index e5084575a4b..e53280d34e5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceProxyTest.java
@@ -45,6 +45,7 @@ public class MetaServiceProxyTest {
private long originReconnectIntervalMs;
private long originRetryCnt;
private boolean originRateLimitEnabled;
+ private boolean originRateLimitDryRun;
private int originRateLimitDefaultQpsPerCore;
private String originRateLimitQpsPerCoreConfig;
private int originRateLimitBurstSeconds;
@@ -56,6 +57,7 @@ public class MetaServiceProxyTest {
originReconnectIntervalMs =
Config.meta_service_rpc_reconnect_interval_ms;
originRetryCnt = Config.meta_service_rpc_retry_cnt;
originRateLimitEnabled = Config.meta_service_rpc_rate_limit_enabled;
+ originRateLimitDryRun = Config.meta_service_rpc_rate_limit_dry_run;
originRateLimitDefaultQpsPerCore =
Config.meta_service_rpc_rate_limit_default_qps_per_core;
originRateLimitQpsPerCoreConfig =
Config.meta_service_rpc_rate_limit_qps_per_core_config;
originRateLimitBurstSeconds =
Config.meta_service_rpc_rate_limit_burst_seconds;
@@ -65,6 +67,7 @@ public class MetaServiceProxyTest {
Config.meta_service_rpc_reconnect_interval_ms = 0;
Config.meta_service_rpc_retry_cnt = 1;
Config.meta_service_rpc_rate_limit_enabled = false;
+ Config.meta_service_rpc_rate_limit_dry_run = false;
Config.meta_service_rpc_rate_limit_default_qps_per_core = 10;
Config.meta_service_rpc_rate_limit_qps_per_core_config = "";
Config.meta_service_rpc_rate_limit_burst_seconds = 1;
@@ -78,6 +81,7 @@ public class MetaServiceProxyTest {
Config.meta_service_rpc_reconnect_interval_ms =
originReconnectIntervalMs;
Config.meta_service_rpc_retry_cnt = originRetryCnt;
Config.meta_service_rpc_rate_limit_enabled = originRateLimitEnabled;
+ Config.meta_service_rpc_rate_limit_dry_run = originRateLimitDryRun;
Config.meta_service_rpc_rate_limit_default_qps_per_core =
originRateLimitDefaultQpsPerCore;
Config.meta_service_rpc_rate_limit_qps_per_core_config =
originRateLimitQpsPerCoreConfig;
Config.meta_service_rpc_rate_limit_burst_seconds =
originRateLimitBurstSeconds;
@@ -524,6 +528,7 @@ public class MetaServiceProxyTest {
private void enableRateLimit(int defaultQpsPerCore, String
qpsPerCoreConfig, int burstSeconds,
long waitTimeoutMs) {
Config.meta_service_rpc_rate_limit_enabled = true;
+ Config.meta_service_rpc_rate_limit_dry_run = false;
Config.meta_service_rpc_rate_limit_default_qps_per_core =
defaultQpsPerCore;
Config.meta_service_rpc_rate_limit_qps_per_core_config =
qpsPerCoreConfig;
Config.meta_service_rpc_rate_limit_burst_seconds = burstSeconds;
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
index e54db40f595..2a31c44618e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/rpc/MetaServiceRpcRateLimiterTest.java
@@ -20,6 +20,12 @@ package org.apache.doris.cloud.rpc;
import org.apache.doris.common.Config;
import org.apache.doris.common.ConfigBase;
import org.apache.doris.common.MetaServiceRpcRateLimitConfigValidator;
+import org.apache.doris.metric.AutoMappedMetric;
+import org.apache.doris.metric.CloudMetrics;
+import org.apache.doris.metric.HistogramMetric;
+import org.apache.doris.metric.LongCounterMetric;
+import org.apache.doris.metric.Metric.MetricUnit;
+import org.apache.doris.metric.MetricRepo;
import org.apache.doris.rpc.RpcException;
import org.junit.After;
@@ -28,26 +34,45 @@ import org.junit.Before;
import org.junit.Test;
import java.lang.reflect.Field;
+import java.util.Collections;
import java.util.concurrent.TimeUnit;
public class MetaServiceRpcRateLimiterTest {
private static final int CPU_CORES =
Runtime.getRuntime().availableProcessors();
private boolean originRateLimitEnabled;
+ private boolean originRateLimitDryRun;
private int originRateLimitDefaultQpsPerCore;
private String originRateLimitQpsPerCoreConfig;
private int originRateLimitBurstSeconds;
private long originRateLimitWaitTimeoutMs;
+ private boolean originMetricRepoIsInit;
+ private AutoMappedMetric<LongCounterMetric> originRateLimitedMetric;
+ private AutoMappedMetric<HistogramMetric> originRateLimitWaitLatencyMetric;
+ private LongCounterMetric originAllRateLimitedMetric;
private MetaServiceRpcRateLimiter rateLimiter;
@Before
public void setUp() {
originRateLimitEnabled = Config.meta_service_rpc_rate_limit_enabled;
+ originRateLimitDryRun = Config.meta_service_rpc_rate_limit_dry_run;
originRateLimitDefaultQpsPerCore =
Config.meta_service_rpc_rate_limit_default_qps_per_core;
originRateLimitQpsPerCoreConfig =
Config.meta_service_rpc_rate_limit_qps_per_core_config;
originRateLimitBurstSeconds =
Config.meta_service_rpc_rate_limit_burst_seconds;
originRateLimitWaitTimeoutMs =
Config.meta_service_rpc_rate_limit_wait_timeout_ms;
+ originMetricRepoIsInit = MetricRepo.isInit;
+ originRateLimitedMetric = CloudMetrics.META_SERVICE_RPC_RATE_LIMITED;
+ originRateLimitWaitLatencyMetric =
CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY;
+ originAllRateLimitedMetric =
CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED;
+
+ CloudMetrics.META_SERVICE_RPC_RATE_LIMITED = new
AutoMappedMetric<>(methodName ->
+ new LongCounterMetric("meta_service_rpc_rate_limited",
MetricUnit.NOUNIT, ""));
+ CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY = new
AutoMappedMetric<>(methodName ->
+ new
HistogramMetric("meta_service_rpc_rate_limit_wait_latency",
Collections.emptyList()));
+ CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED = new LongCounterMetric(
+ "meta_service_rpc_all_rate_limited", MetricUnit.NOUNIT, "");
+ MetricRepo.isInit = true;
rateLimiter = new MetaServiceRpcRateLimiter();
enableRateLimit(1, "", 1, 0);
@@ -56,10 +81,15 @@ public class MetaServiceRpcRateLimiterTest {
@After
public void tearDown() {
Config.meta_service_rpc_rate_limit_enabled = originRateLimitEnabled;
+ Config.meta_service_rpc_rate_limit_dry_run = originRateLimitDryRun;
Config.meta_service_rpc_rate_limit_default_qps_per_core =
originRateLimitDefaultQpsPerCore;
Config.meta_service_rpc_rate_limit_qps_per_core_config =
originRateLimitQpsPerCoreConfig;
Config.meta_service_rpc_rate_limit_burst_seconds =
originRateLimitBurstSeconds;
Config.meta_service_rpc_rate_limit_wait_timeout_ms =
originRateLimitWaitTimeoutMs;
+ CloudMetrics.META_SERVICE_RPC_RATE_LIMITED = originRateLimitedMetric;
+ CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY =
originRateLimitWaitLatencyMetric;
+ CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED =
originAllRateLimitedMetric;
+ MetricRepo.isInit = originMetricRepoIsInit;
rateLimiter.reset();
}
@@ -81,6 +111,34 @@ public class MetaServiceRpcRateLimiterTest {
rateLimiter.acquire("disabledSwitch");
}
+ @Test
+ public void testDryRunDoesNotReject() throws RpcException {
+ Config.meta_service_rpc_rate_limit_dry_run = true;
+ consumePermits("dryRun", CPU_CORES);
+
+ rateLimiter.acquire("dryRun");
+ Assert.assertEquals(1L,
CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED.getValue().longValue());
+ Assert.assertEquals(1L,
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMITED.getOrAdd("dryRun").getValue().longValue());
+
+ Config.meta_service_rpc_rate_limit_dry_run = false;
+ assertRateLimited("dryRun");
+ Assert.assertEquals(2L,
CloudMetrics.META_SERVICE_RPC_ALL_RATE_LIMITED.getValue().longValue());
+ Assert.assertEquals(2L,
+
CloudMetrics.META_SERVICE_RPC_RATE_LIMITED.getOrAdd("dryRun").getValue().longValue());
+ }
+
+ @Test
+ public void testDryRunDoesNotWait() throws RpcException {
+ enableRateLimit(1, "", 1, 2000);
+ Config.meta_service_rpc_rate_limit_dry_run = true;
+ consumePermits("dryRunWait", CPU_CORES);
+
+ Assert.assertEquals(0, rateLimiter.acquire("dryRunWait"));
+ Assert.assertEquals(1L,
CloudMetrics.META_SERVICE_RPC_RATE_LIMIT_WAIT_LATENCY.getOrAdd("dryRunWait")
+ .getHistogram().getCount());
+ }
+
@Test
public void testMethodOverrideQpsTakesEffect() throws RpcException {
enableRateLimit(1, "fast:2", 1, 0);
@@ -233,6 +291,7 @@ public class MetaServiceRpcRateLimiterTest {
private void enableRateLimit(int defaultQpsPerCore, String
qpsPerCoreConfig, int burstSeconds,
long waitTimeoutMs) {
Config.meta_service_rpc_rate_limit_enabled = true;
+ Config.meta_service_rpc_rate_limit_dry_run = false;
Config.meta_service_rpc_rate_limit_default_qps_per_core =
defaultQpsPerCore;
Config.meta_service_rpc_rate_limit_qps_per_core_config =
qpsPerCoreConfig;
Config.meta_service_rpc_rate_limit_burst_seconds = burstSeconds;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
