feature: add single server rate limit (#6756)

* feature: add single server rale limit * fix: fix RateLimitInfo * fix: fix ResultCodeTest * fix: fix TokenBucketLimiterTest * fix: fix pmd output * fix: fix pmd output * fix: fix pmd output * fix: fix pmd * fix: add relate config * fix: fix bucketTokenNumPerSecond config * feat: add ratelimit hot config and del counter metrics * feat: add ratelimit hot config and del counter metrics * feat: add RateLimiterHandlerConfig license * fix: fix MockFailureHandlerImpl * fix: fix pom.xml pmd * fix: fix pmd check * fix: fix pmd check * fix: fix errer code in client * fix: fix tm modify * fix: fix tm modify * fix: fix as review by funky * fix: add limit chain * style: fix import code style * style: fix import code style * style: fix pmd check * fix: fix as code review --------- Co-authored-by: Lei Zhiyuan <[email protected]> Co-authored-by: jimin <[email protected]> Co-authored-by: funkye <[email protected]> Co-authored-by: xingfudeshi <[email protected]>
apache · Jan 22, 2025 · 4ed95e4 · 4ed95e4
1 parent ab4574e
commit 4ed95e4
Show file tree

Hide file tree

Showing 37 changed files with 1,125 additions and 14 deletions.
diff --git a/build/pom.xml b/build/pom.xml
@@ -89,6 +89,7 @@
         <kafka-clients.version>3.6.1</kafka-clients.version>
         <snakeyaml.version>2.0</snakeyaml.version>
         <jackson-mapper.version>1.9.13</jackson-mapper.version>
+        <bucket4j.version>8.1.0</bucket4j.version>
 
         <!-- For test -->
         <junit-jupiter.version>5.8.2</junit-jupiter.version>

diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
@@ -5,6 +5,7 @@ Add changes here for all PR submitted to the 2.x branch.
 ### feature:
 
 - [[#7073](https://github.com/apache/incubator-seata/pull/7073)] support virtual thread,replace the usages of synchronized with ReentrantLock
+- [[#6756](https://github.com/apache/incubator-seata/pull/6756)] feature: add single server rate limit
 - [[#7037](https://github.com/apache/incubator-seata/pull/7037)] support fury undolog parser
 - [[#7069](https://github.com/apache/incubator-seata/pull/7069)] Raft cluster mode supports address translation
 - [[#7038](https://github.com/apache/incubator-seata/pull/7038)] support fury serializer
@@ -49,6 +50,7 @@ Thanks to these contributors for their code commits. Please report an unintended
 - [lightClouds917](https://github.com/lightClouds917)
 - [GoodBoyCoder](https://github.com/GoodBoyCoder)
 - [PeppaO](https://github.com/PeppaO)
+- [xjlgod](https://github.com/xjlgod)
 - [funky-eyes](https://github.com/funky-eyes)
 - [psxjoy](https://github.com/psxjoy)
 - [xiaoxiangyeyu0](https://github.com/xiaoxiangyeyu0)

diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
@@ -4,6 +4,7 @@
 
 ### feature:
 
+- [[#6756](https://github.com/apache/incubator-seata/pull/6756)] seata服务单点限流支持
 - [[#7073](https://github.com/apache/incubator-seata/pull/7073)] 支持虚拟线程，用ReentrantLock替换synchronized的用法
 - [[#7037](https://github.com/apache/incubator-seata/pull/7037)] 支持UndoLog的fury序列化方式
 - [[#7069](https://github.com/apache/incubator-seata/pull/7069)] Raft集群模式支持地址转换
@@ -45,6 +46,7 @@
 - [slievrly](https://github.com/slievrly)
 - [lyl2008dsg](https://github.com/lyl2008dsg)
 - [remind](https://github.com/remind)
+- [xjlgod](https://github.com/xjlgod)
 - [lightClouds917](https://github.com/lightClouds917)
 - [GoodBoyCoder](https://github.com/GoodBoyCoder)
 - [PeppaO](https://github.com/PeppaO)

diff --git a/common/src/main/java/org/apache/seata/common/ConfigurationKeys.java b/common/src/main/java/org/apache/seata/common/ConfigurationKeys.java
@@ -1121,6 +1121,31 @@ public interface ConfigurationKeys {
      */
     String META_PREFIX = SEATA_FILE_ROOT_CONFIG + FILE_CONFIG_SPLIT_CHAR + FILE_ROOT_REGISTRY + FILE_CONFIG_SPLIT_CHAR + "metadata.";
 
+    /**
+     * The constant RATE_LIMIT_PREFIX.
+     */
+    String RATE_LIMIT_PREFIX = SERVER_PREFIX + "ratelimit";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_NUM_PER_SECOND = RATE_LIMIT_PREFIX + ".bucketTokenNumPerSecond";
+
+    /**
+     * The constant RATE_LIMIT_ENABLE.
+     */
+    String RATE_LIMIT_ENABLE = RATE_LIMIT_PREFIX + ".enable";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_MAX_NUM.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_MAX_NUM = RATE_LIMIT_PREFIX + ".bucketTokenMaxNum";
+
+    /**
+     * The constant RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM.
+     */
+    String RATE_LIMIT_BUCKET_TOKEN_INITIAL_NUM = RATE_LIMIT_PREFIX + ".bucketTokenInitialNum";
+
     /**
      * The constant SERVER_REGISTRY_METADATA_PREFIX
      */

diff --git a/common/src/main/java/org/apache/seata/common/DefaultValues.java b/common/src/main/java/org/apache/seata/common/DefaultValues.java
@@ -497,6 +497,12 @@ public interface DefaultValues {
      */
     int DEFAULT_ROCKET_MQ_MSG_TIMEOUT = 60 * 1000;
 
+    /**
+     * The constant DEFAULT_RATE_LIMIT_ENABLE.
+     */
+    boolean DEFAULT_RATE_LIMIT_ENABLE = false;
+
+
     /**
      * The constant DEFAULT_RAFT_SSL_ENABLED.
      */

diff --git a/compatible/src/main/java/io/seata/tm/api/DefaultFailureHandlerImpl.java b/compatible/src/main/java/io/seata/tm/api/DefaultFailureHandlerImpl.java
@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+
 /**
  * The type Default failure handler.
  */

diff --git a/core/src/main/java/org/apache/seata/core/event/RateLimitEvent.java b/core/src/main/java/org/apache/seata/core/event/RateLimitEvent.java
@@ -0,0 +1,104 @@
+/*
+ * 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.seata.core.event;
+
+public class RateLimitEvent implements Event {
+
+    /**
+     * The Trace id.
+     */
+    private String traceId;
+
+    /**
+     * The Limit type (like GlobalBeginFailed).
+     */
+    private String limitType;
+
+    /**
+     * The Application id.
+     */
+    private String applicationId;
+
+    /**
+     * The Client id.
+     */
+    private String clientId;
+
+    /**
+     * The Server ip address and port.
+     */
+    private String serverIpAddressAndPort;
+
+    public String getTraceId() {
+        return traceId;
+    }
+
+    public void setTraceId(String traceId) {
+        this.traceId = traceId;
+    }
+
+    public String getLimitType() {
+        return limitType;
+    }
+
+    public void setLimitType(String limitType) {
+        this.limitType = limitType;
+    }
+
+    public String getApplicationId() {
+        return applicationId;
+    }
+
+    public void setApplicationId(String applicationId) {
+        this.applicationId = applicationId;
+    }
+
+    public String getClientId() {
+        return clientId;
+    }
+
+    public void setClientId(String clientId) {
+        this.clientId = clientId;
+    }
+
+    public String getServerIpAddressAndPort() {
+        return serverIpAddressAndPort;
+    }
+
+    public void setServerIpAddressAndPort(String serverIpAddressAndPort) {
+        this.serverIpAddressAndPort = serverIpAddressAndPort;
+    }
+
+    public RateLimitEvent(String traceId, String limitType, String applicationId, String clientId, String serverIpAddressAndPort) {
+        this.traceId = traceId;
+        this.limitType = limitType;
+        this.applicationId = applicationId;
+        this.clientId = clientId;
+        this.serverIpAddressAndPort = serverIpAddressAndPort;
+    }
+
+    @Override
+    public String toString() {
+        return "RateLimitEvent{" +
+                "traceId='" + traceId + '\'' +
+                ", limitType='" + limitType + '\'' +
+                ", applicationId='" + applicationId + '\'' +
+                ", clientId='" + clientId + '\'' +
+                ", serverIpAddressAndPort='" + serverIpAddressAndPort + '\'' +
+                '}';
+    }
+}
diff --git a/core/src/main/java/org/apache/seata/core/exception/TransactionExceptionCode.java b/core/src/main/java/org/apache/seata/core/exception/TransactionExceptionCode.java
@@ -31,7 +31,6 @@ public enum TransactionExceptionCode {
      * BeginFailed
      */
     BeginFailed,
-
     /**
      * Lock key conflict transaction exception code.
      */

diff --git a/core/src/main/java/org/apache/seata/core/protocol/MessageType.java b/core/src/main/java/org/apache/seata/core/protocol/MessageType.java
@@ -22,6 +22,11 @@
  */
 public interface MessageType {
 
+    /**
+     * The constant TYPE_NOT_EXIST.
+     */
+    short TYPE_NOT_EXIST = 0;
+
     /**
      * The constant TYPE_GLOBAL_BEGIN.
      */

diff --git a/core/src/test/java/org/apache/seata/core/protocol/ResultCodeTest.java b/core/src/test/java/org/apache/seata/core/protocol/ResultCodeTest.java
@@ -45,7 +45,8 @@ void getInt() {
 
     @Test
     void values() {
-        Assertions.assertArrayEquals(new ResultCode[]{ResultCode.Failed, ResultCode.Success}, ResultCode.values());
+        Assertions.assertArrayEquals(new ResultCode[]{ResultCode.Failed, ResultCode.Success},
+                ResultCode.values());
     }
 
     @Test

diff --git a/dependencies/pom.xml b/dependencies/pom.xml
@@ -68,6 +68,7 @@
         <sofa.registry.version>6.3.0</sofa.registry.version>
         <motan.version>1.0.0</motan.version>
         <jcommander.version>1.82</jcommander.version>
+        <bucket4j.version>8.1.0</bucket4j.version>
         <commons-compress.version>1.21</commons-compress.version>
         <ant.version>1.10.12</ant.version>
         <lz4.version>1.7.1</lz4.version>
@@ -618,6 +619,11 @@
                 <version>${jcommander.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>com.bucket4j</groupId>
+                <artifactId>bucket4j_jdk8-core</artifactId>
+                <version>${bucket4j.version}</version>
+            </dependency>
             <dependency>
                 <groupId>io.grpc</groupId>
                 <artifactId>grpc-testing</artifactId>

diff --git a/...e/seata/integration/tx/api/interceptor/handler/GlobalTransactionalInterceptorHandler.java b/...e/seata/integration/tx/api/interceptor/handler/GlobalTransactionalInterceptorHandler.java
@@ -56,6 +56,7 @@
 import org.apache.seata.tm.api.transaction.NoRollbackRule;
 import org.apache.seata.tm.api.transaction.RollbackRule;
 import org.apache.seata.tm.api.transaction.TransactionInfo;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

diff --git a/metrics/seata-metrics-api/src/main/java/org/apache/seata/metrics/IdConstants.java b/metrics/seata-metrics-api/src/main/java/org/apache/seata/metrics/IdConstants.java
@@ -25,6 +25,8 @@ public interface IdConstants {
 
     String SEATA_EXCEPTION = "seata.exception";
 
+    String SEATA_RATE_LIMIT = "seata.rate.limit";
+
     String APP_ID_KEY = "applicationId";
 
     String GROUP_KEY = "group";
@@ -79,4 +81,9 @@ public interface IdConstants {
 
     String STATUS_VALUE_AFTER_ROLLBACKED_KEY = "AfterRollbacked";
 
+    String LIMIT_TYPE_KEY = "limitType";
+
+    String CLIENT_ID_KEY = "clientId";
+
+    String HOST_AND_PORT = "hostAndPort";
 }
diff --git a/...pring/src/main/java/org/apache/seata/saga/engine/tm/DefaultSagaTransactionalTemplate.java b/...pring/src/main/java/org/apache/seata/saga/engine/tm/DefaultSagaTransactionalTemplate.java
@@ -16,15 +16,14 @@
  */
 package org.apache.seata.saga.engine.tm;
 
-import java.util.List;
-
 import org.apache.seata.common.exception.FrameworkErrorCode;
+import org.apache.seata.common.util.StringUtils;
 import org.apache.seata.core.exception.TransactionException;
 import org.apache.seata.core.model.BranchStatus;
 import org.apache.seata.core.model.BranchType;
 import org.apache.seata.core.model.GlobalStatus;
-import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
 import org.apache.seata.core.rpc.ShutdownHook;
+import org.apache.seata.core.rpc.netty.RmNettyRemotingClient;
 import org.apache.seata.core.rpc.netty.TmNettyRemotingClient;
 import org.apache.seata.rm.DefaultResourceManager;
 import org.apache.seata.rm.RMClient;
@@ -39,7 +38,6 @@
 import org.apache.seata.tm.api.transaction.TransactionHook;
 import org.apache.seata.tm.api.transaction.TransactionHookManager;
 import org.apache.seata.tm.api.transaction.TransactionInfo;
-import org.apache.seata.common.util.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -49,6 +47,8 @@
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.ConfigurableApplicationContext;
 
+import java.util.List;
+
 /**
  * Template of executing business logic with a global transaction for SAGA mode
  */
@@ -93,7 +93,6 @@ public GlobalTransaction beginTransaction(TransactionInfo txInfo) throws Transac
             triggerAfterBegin(tx);
         } catch (TransactionException txe) {
             throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure);
-
         }
         return tx;
     }

diff --git a/script/config-center/config.txt b/script/config-center/config.txt
@@ -162,7 +162,10 @@ server.raft.serialization=jackson
 server.raft.compressor=none
 server.raft.sync=true
 
-
+server.ratelimit.enable=false
+server.ratelimit.bucketTokenNumPerSecond = 999999
+server.ratelimit.bucketTokenMaxNum = 999999
+server.ratelimit.bucketTokenInitialNum = 999999
 
 #Metrics configuration, only for the server
 metrics.enabled=true

diff --git a/...igure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/StarterConstants.java b/...igure-core/src/main/java/org/apache/seata/spring/boot/autoconfigure/StarterConstants.java
@@ -71,6 +71,7 @@ public interface StarterConstants {
 
 
     String SERVER_PREFIX = SEATA_PREFIX + ".server";
+    String SERVER_RATELIMIT_PREFIX = SERVER_PREFIX + ".ratelimit";
     String SERVER_UNDO_PREFIX = SERVER_PREFIX + ".undo";
     String SERVER_RAFT_PREFIX = SERVER_PREFIX + ".raft";
     String SERVER_RAFT_SSL_PREFIX = SERVER_RAFT_PREFIX + ".ssl";

diff --git a/.../java/org/apache/seata/spring/boot/autoconfigure/SeataServerEnvironmentPostProcessor.java b/.../java/org/apache/seata/spring/boot/autoconfigure/SeataServerEnvironmentPostProcessor.java
@@ -17,6 +17,8 @@
 package org.apache.seata.spring.boot.autoconfigure;
 
 import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.seata.spring.boot.autoconfigure.properties.server.ServerRateLimitProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.raft.ServerRaftSSLClientProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.raft.ServerRaftSSLProperties;
 import org.apache.seata.spring.boot.autoconfigure.properties.server.raft.ServerRaftSSLServerProperties;
@@ -41,6 +43,7 @@
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.PROPERTY_BEAN_MAP;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RAFT_PREFIX;
+import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RATELIMIT_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RAFT_SSL_CLIENT_KEYSTORE_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RAFT_SSL_PREFIX;
 import static org.apache.seata.spring.boot.autoconfigure.StarterConstants.SERVER_RAFT_SSL_SERVER_KEYSTORE_PREFIX;
@@ -90,6 +93,7 @@ public static void init() {
             PROPERTY_BEAN_MAP.put(SERVER_RAFT_SSL_CLIENT_KEYSTORE_PREFIX, ServerRaftSSLClientProperties.class);
             PROPERTY_BEAN_MAP.put(SESSION_PREFIX, SessionProperties.class);
             PROPERTY_BEAN_MAP.put(STORE_PREFIX, StoreProperties.class);
+            PROPERTY_BEAN_MAP.put(SERVER_RATELIMIT_PREFIX, ServerRateLimitProperties.class);
         }
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -28,6 +28,7 @@ @@
     import org.slf4j.Logger;
     import org.slf4j.LoggerFactory;
     /**
      * The type Default failure handler.
      */
@@ Expand Down @@