Add attributes of cluster concurrency limiting in ClusterFlowConfig

Signed-off-by: yunfeiyanggzq <yunfeiyang@buaa.edu.cn>

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/RuleConstant.java

@@ -51,6 +51,14 @@ public final class RuleConstant {
     public static final int CONTROL_BEHAVIOR_RATE_LIMITER = 2;
     public static final int CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER = 3;
 
+    public static final int DEFAULT_BLOCK_STRATEGY = 0;
+    public static final int TRY_AGAIN_BLOCK_STRATEGY = 1;
+    public static final int TRY_UNTIL_SUCCESS_BLOCK_STRATEGY = 2;
+
+    public static final int DEFAULT_RESOURCE_TIMEOUT_STRATEGY = 0;
+    public static final int RELEASE_RESOURCE_TIMEOUT_STRATEGY = 1;
+    public static final int KEEP_RESOURCE_TIMEOUT_STRATEGY = 2;
+
     public static final String LIMIT_APP_DEFAULT = "default";
     public static final String LIMIT_APP_OTHER = "other";
 

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/ClusterFlowConfig.java

@@ -18,6 +18,8 @@ package com.alibaba.csp.sentinel.slots.block.flow;
 import com.alibaba.csp.sentinel.slots.block.ClusterRuleConstant;
 import com.alibaba.csp.sentinel.slots.block.RuleConstant;
 
+import java.util.Objects;
+
 /**
  * Flow rule config in cluster mode.
  *
@@ -48,6 +50,61 @@ public class ClusterFlowConfig {
      */
     private int windowIntervalMs = RuleConstant.DEFAULT_WINDOW_INTERVAL_MS;
 
+    /**
+     * if the client keep the token for more than resourceTimeout,resourceTimeoutStrategy will work.
+     */
+    private long resourceTimeout = 2000;
+
+    /**
+     * 0:ignore,1:release the token.
+     */
+    private int resourceTimeoutStrategy = RuleConstant.DEFAULT_RESOURCE_TIMEOUT_STRATEGY;
+
+    /**
+     * if the request(prioritized=true) is block,acquireRefuseStrategy will work..
+     * 0:ignore and block.
+     * 1:try again .
+     * 2:try until success.
+     */
+    private int acquireRefuseStrategy = RuleConstant.DEFAULT_BLOCK_STRATEGY;
+
+    /**
+     * if a client is offline,the server will delete all the token the client holds after clientOfflineTime.
+     */
+    private long clientOfflineTime = 2000;
+
+    public long getResourceTimeout() {
+        return resourceTimeout;
+    }
+
+    public void setResourceTimeout(long resourceTimeout) {
+        this.resourceTimeout = resourceTimeout;
+    }
+
+    public int getResourceTimeoutStrategy() {
+        return resourceTimeoutStrategy;
+    }
+
+    public void setResourceTimeoutStrategy(int resourceTimeoutStrategy) {
+        this.resourceTimeoutStrategy = resourceTimeoutStrategy;
+    }
+
+    public int getAcquireRefuseStrategy() {
+        return acquireRefuseStrategy;
+    }
+
+    public void setAcquireRefuseStrategy(int acquireRefuseStrategy) {
+        this.acquireRefuseStrategy = acquireRefuseStrategy;
+    }
+
+    public long getClientOfflineTime() {
+        return clientOfflineTime;
+    }
+
+    public void setClientOfflineTime(long clientOfflineTime) {
+        this.clientOfflineTime = clientOfflineTime;
+    }
+
     public Long getFlowId() {
         return flowId;
     }
@@ -104,17 +161,43 @@ public class ClusterFlowConfig {
 
     @Override
     public boolean equals(Object o) {
-        if (this == o) { return true; }
-        if (o == null || getClass() != o.getClass()) { return false; }
-
-        ClusterFlowConfig that = (ClusterFlowConfig)o;
-
-        if (thresholdType != that.thresholdType) { return false; }
-        if (fallbackToLocalWhenFail != that.fallbackToLocalWhenFail) { return false; }
-        if (strategy != that.strategy) { return false; }
-        if (sampleCount != that.sampleCount) { return false; }
-        if (windowIntervalMs != that.windowIntervalMs) { return false; }
-        return flowId != null ? flowId.equals(that.flowId) : that.flowId == null;
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        ClusterFlowConfig that = (ClusterFlowConfig) o;
+
+        if (thresholdType != that.thresholdType) {
+            return false;
+        }
+        if (fallbackToLocalWhenFail != that.fallbackToLocalWhenFail) {
+            return false;
+        }
+        if (strategy != that.strategy) {
+            return false;
+        }
+        if (sampleCount != that.sampleCount) {
+            return false;
+        }
+        if (windowIntervalMs != that.windowIntervalMs) {
+            return false;
+        }
+        if (resourceTimeout != that.resourceTimeout) {
+            return false;
+        }
+        if (clientOfflineTime != that.clientOfflineTime) {
+            return false;
+        }
+        if (resourceTimeoutStrategy != that.resourceTimeoutStrategy) {
+            return false;
+        }
+        if (acquireRefuseStrategy != that.acquireRefuseStrategy) {
+            return false;
+        }
+        return Objects.equals(flowId, that.flowId);
     }
 
     @Override
@@ -125,18 +208,26 @@ public class ClusterFlowConfig {
         result = 31 * result + strategy;
         result = 31 * result + sampleCount;
         result = 31 * result + windowIntervalMs;
+        result = (int) (31 * result + resourceTimeout);
+        result = (int) (31 * result + clientOfflineTime);
+        result = 31 * result + resourceTimeoutStrategy;
+        result = 31 * result + acquireRefuseStrategy;
         return result;
     }
 
     @Override
     public String toString() {
         return "ClusterFlowConfig{" +
-            "flowId=" + flowId +
-            ", thresholdType=" + thresholdType +
-            ", fallbackToLocalWhenFail=" + fallbackToLocalWhenFail +
-            ", strategy=" + strategy +
-            ", sampleCount=" + sampleCount +
-            ", windowIntervalMs=" + windowIntervalMs +
-            '}';
+                "flowId=" + flowId +
+                ", thresholdType=" + thresholdType +
+                ", fallbackToLocalWhenFail=" + fallbackToLocalWhenFail +
+                ", strategy=" + strategy +
+                ", sampleCount=" + sampleCount +
+                ", windowIntervalMs=" + windowIntervalMs +
+                ", resourceTimeout=" + resourceTimeout +
+                ", resourceTimeoutStrategy=" + resourceTimeoutStrategy +
+                ", acquireRefuseStrategy=" + acquireRefuseStrategy +
+                ", clientOfflineTime=" + clientOfflineTime +
+                '}';
     }
 }

sentinel-core/src/main/java/com/alibaba/csp/sentinel/slots/block/flow/FlowRuleUtil.java

@@ -15,16 +15,6 @@
  */
 package com.alibaba.csp.sentinel.slots.block.flow;
 
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
 import com.alibaba.csp.sentinel.log.RecordLog;
 import com.alibaba.csp.sentinel.slots.block.ClusterRuleConstant;
 import com.alibaba.csp.sentinel.slots.block.RuleConstant;
@@ -36,6 +26,10 @@ import com.alibaba.csp.sentinel.util.StringUtil;
 import com.alibaba.csp.sentinel.util.function.Function;
 import com.alibaba.csp.sentinel.util.function.Predicate;
 
+import java.util.*;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+
 /**
  * @author Eric Zhao
  * @since 1.4.0
@@ -55,8 +49,8 @@ public final class FlowRuleUtil {
     /**
      * Build the flow rule map from raw list of flow rules, grouping by resource name.
      *
-     * @param list          raw list of flow rules
-     * @param filter        rule filter
+     * @param list   raw list of flow rules
+     * @param filter rule filter
      * @return constructed new flow rule map; empty map if list is null or empty, or no wanted rules
      */
     public static Map<String, List<FlowRule>> buildFlowRuleMap(List<FlowRule> list, Predicate<FlowRule> filter) {
@@ -66,9 +60,9 @@ public final class FlowRuleUtil {
     /**
      * Build the flow rule map from raw list of flow rules, grouping by resource name.
      *
-     * @param list          raw list of flow rules
-     * @param filter        rule filter
-     * @param shouldSort    whether the rules should be sorted
+     * @param list       raw list of flow rules
+     * @param filter     rule filter
+     * @param shouldSort whether the rules should be sorted
      * @return constructed new flow rule map; empty map if list is null or empty, or no wanted rules
      */
     public static Map<String, List<FlowRule>> buildFlowRuleMap(List<FlowRule> list, Predicate<FlowRule> filter,
@@ -105,7 +99,6 @@ public final class FlowRuleUtil {
             if (StringUtil.isBlank(rule.getLimitApp())) {
                 rule.setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
             }
-
             TrafficShapingController rater = generateRater(rule);
             rule.setRater(rater);
 
@@ -141,12 +134,12 @@ public final class FlowRuleUtil {
             switch (rule.getControlBehavior()) {
                 case RuleConstant.CONTROL_BEHAVIOR_WARM_UP:
                     return new WarmUpController(rule.getCount(), rule.getWarmUpPeriodSec(),
-                        ColdFactorProperty.coldFactor);
+                            ColdFactorProperty.coldFactor);
                 case RuleConstant.CONTROL_BEHAVIOR_RATE_LIMITER:
                     return new RateLimiterController(rule.getMaxQueueingTimeMs(), rule.getCount());
                 case RuleConstant.CONTROL_BEHAVIOR_WARM_UP_RATE_LIMITER:
                     return new WarmUpRateLimiterController(rule.getCount(), rule.getWarmUpPeriodSec(),
-                        rule.getMaxQueueingTimeMs(), ColdFactorProperty.coldFactor);
+                            rule.getMaxQueueingTimeMs(), ColdFactorProperty.coldFactor);
                 case RuleConstant.CONTROL_BEHAVIOR_DEFAULT:
                 default:
                     // Default mode or unknown mode: default traffic shaping controller (fast-reject).
@@ -173,12 +166,42 @@ public final class FlowRuleUtil {
      */
     public static boolean isValidRule(FlowRule rule) {
         boolean baseValid = rule != null && !StringUtil.isBlank(rule.getResource()) && rule.getCount() >= 0
-            && rule.getGrade() >= 0 && rule.getStrategy() >= 0 && rule.getControlBehavior() >= 0;
+                && rule.getGrade() >= 0 && rule.getStrategy() >= 0 && rule.getControlBehavior() >= 0;
         if (!baseValid) {
             return false;
         }
-        // Check strategy and control (shaping) behavior.
-        return checkClusterField(rule) && checkStrategyField(rule) && checkControlBehaviorField(rule);
+        if (rule.getGrade() == RuleConstant.FLOW_GRADE_QPS) {
+            // Check strategy and control (shaping) behavior.
+            return checkClusterField(rule) && checkStrategyField(rule) && checkControlBehaviorField(rule);
+        } else if (rule.getGrade() == RuleConstant.FLOW_GRADE_THREAD) {
+            return checkClusterConcurrentField(rule);
+        } else {
+            return false;
+        }
+
+    }
+
+    public static boolean checkClusterConcurrentField(/*@NonNull*/ FlowRule rule) {
+        if (!rule.isClusterMode()) {
+            return true;
+        }
+        ClusterFlowConfig clusterConfig = rule.getClusterConfig();
+        if (clusterConfig == null) {
+            return false;
+        }
+        if (clusterConfig.getClientOfflineTime() <= 0 || clusterConfig.getResourceTimeout() <= 0) {
+            return false;
+        }
+
+        if (clusterConfig.getAcquireRefuseStrategy() < 0 || clusterConfig.getResourceTimeoutStrategy() < 0) {
+            return false;
+        }
+
+        if (!validClusterRuleId(clusterConfig.getFlowId())) {
+            return false;
+        }
+
+        return isWindowConfigValid(clusterConfig.getSampleCount(), clusterConfig.getWindowIntervalMs());
     }
 
     private static boolean checkClusterField(/*@NonNull*/ FlowRule rule) {
@@ -234,5 +257,6 @@ public final class FlowRuleUtil {
         }
     };
 
-    private FlowRuleUtil() {}
+    private FlowRuleUtil() {
+    }
 }