Add Xds flow control service

Signed-off-by: hanbingleixue <[email protected]>

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/XdsCoreService.java

@@ -45,4 +45,11 @@ public interface XdsCoreService extends BaseService {
      * @return XdsRoute
      */
     XdsLoadBalanceService getLoadBalanceService();
+
+    /**
+     * get XDSFlowControlService
+     *
+     * @return XdsFlowControlService
+     */
+    XdsFlowControlService getXdsFlowControlService();
 }

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/XdsFlowControlService.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright (C) 2024-2024 Sermant Authors. All rights reserved.
+ *
+ *   Licensed 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 io.sermant.core.service.xds;
+
+import io.sermant.core.service.xds.entity.XdsHttpFault;
+import io.sermant.core.service.xds.entity.XdsInstanceCircuitBreakers;
+import io.sermant.core.service.xds.entity.XdsRateLimit;
+import io.sermant.core.service.xds.entity.XdsRequestCircuitBreakers;
+import io.sermant.core.service.xds.entity.XdsRetryPolicy;
+
+import java.util.Optional;
+
+/**
+ * xDS FlowControl service
+ *
+ * @author zhp
+ * @since 2024-11-27
+ **/
+public interface XdsFlowControlService {
+    /**
+     * get request circuit breaker information of cluster
+     *
+     * @param serviceName service name
+     * @param clusterName cluster name
+     * @return circuit breaker rules
+     */
+    Optional<XdsRequestCircuitBreakers> getRequestCircuitBreakers(String serviceName, String clusterName);
+
+    /**
+     * get instance circuit breaker information of cluster
+     *
+     * @param serviceName service name
+     * @param clusterName cluster name
+     * @return Outlier Detection rules
+     */
+    Optional<XdsInstanceCircuitBreakers> getInstanceCircuitBreakers(String serviceName, String clusterName);
+
+    /**
+     * get retry policy of route name
+     *
+     * @param serviceName service name
+     * @param routeName route name
+     * @return retry policy
+     */
+    Optional<XdsRetryPolicy> getRetryPolicy(String serviceName, String routeName);
+
+    /**
+     * get rate limit of route name
+     *
+     * @param serviceName service name
+     * @param routeName route name
+     * @return rate limit rule
+     */
+    Optional<XdsRateLimit> getRateLimit(String serviceName, String routeName);
+
+    /**
+     * get http fault of route name
+     *
+     * @param serviceName service name
+     * @param routeName route name
+     * @return http fault rule
+     */
+    Optional<XdsHttpFault> getHttpFault(String serviceName, String routeName);
+}

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsAbort.java

@@ -31,7 +31,7 @@ public class XdsAbort {
     /**
      * The percentage of requests/ operations/ connections that will be aborted with the error code
      */
-    private float percentage;
+    private int percentage;
 
     public int getHttpStatus() {
         return httpStatus;
@@ -41,11 +41,11 @@ public void setHttpStatus(int httpStatus) {
         this.httpStatus = httpStatus;
     }
 
-    public float getPercentage() {
+    public int getPercentage() {
         return percentage;
     }
 
-    public void setPercentage(float percentage) {
+    public void setPercentage(int percentage) {
         this.percentage = percentage;
     }
 }

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsCluster.java

@@ -31,6 +31,10 @@ public class XdsCluster {
 
     private boolean isLocalityLb;
 
+    private XdsRequestCircuitBreakers requestCircuitBreakers;
+
+    private XdsInstanceCircuitBreakers instanceCircuitBreakers;
+
     public String getClusterName() {
         return clusterName;
     }
@@ -62,4 +66,20 @@ public boolean isLocalityLb() {
     public void setLocalityLb(boolean localityLb) {
         isLocalityLb = localityLb;
     }
+
+    public XdsRequestCircuitBreakers getRequestCircuitBreakers() {
+        return requestCircuitBreakers;
+    }
+
+    public void setRequestCircuitBreakers(XdsRequestCircuitBreakers requestCircuitBreakers) {
+        this.requestCircuitBreakers = requestCircuitBreakers;
+    }
+
+    public XdsInstanceCircuitBreakers getInstanceCircuitBreakers() {
+        return instanceCircuitBreakers;
+    }
+
+    public void setInstanceCircuitBreakers(XdsInstanceCircuitBreakers instanceCircuitBreakers) {
+        this.instanceCircuitBreakers = instanceCircuitBreakers;
+    }
 }

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsDelay.java

@@ -31,7 +31,7 @@ public class XdsDelay {
     /**
      * The percentage of requests on which the delay will be injected
      */
-    private float percentage;
+    private int percentage;
 
     public long getFixedDelay() {
         return fixedDelay;
@@ -41,11 +41,11 @@ public void setFixedDelay(long fixedDelay) {
         this.fixedDelay = fixedDelay;
     }
 
-    public float getPercentage() {
+    public int getPercentage() {
         return percentage;
     }
 
-    public void setPercentage(float percentage) {
+    public void setPercentage(int percentage) {
         this.percentage = percentage;
     }
 }

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsOutlierDetection.java → sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsInstanceCircuitBreakers.java

@@ -17,12 +17,13 @@
 package io.sermant.core.service.xds.entity;
 
 /**
- * Xds Outlier Detection information
+ * Circuit breaker information for downstream instance, The instance has reached the specified number of errors and will
+ * trigger a circuit breaker and the instance will be removed for a period of time
  *
  * @author zhp
  * @since 2024-11-18
  */
-public class XdsOutlierDetection {
+public class XdsInstanceCircuitBreakers {
     /**
      * Whether to distinguish between local source failures and external errors. When set to true,
      * it will detect local source failures.

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsCircuitBreakers.java → sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsRequestCircuitBreakers.java

@@ -17,12 +17,13 @@
 package io.sermant.core.service.xds.entity;
 
 /**
- * Xds circuit breaker information
+ * Circuit breaker information for downstream instance requests, When the number of active requests for an instance
+ * reaches the specified limit, it will trigger a circuit breaker
  *
  * @author zhp
  * @since 2024-11-18
  */
-public class XdsCircuitBreakers {
+public class XdsRequestCircuitBreakers {
     /**
      * Maximum active request count
      */

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsRoute.java

@@ -29,6 +29,10 @@ public class XdsRoute {
 
     private XdsRouteAction routeAction;
 
+    private XdsHttpFault httpFault;
+
+    private XdsRateLimit rateLimit;
+
     public String getName() {
         return name;
     }
@@ -52,4 +56,20 @@ public XdsRouteAction getRouteAction() {
     public void setRouteAction(XdsRouteAction routeAction) {
         this.routeAction = routeAction;
     }
+
+    public XdsHttpFault getHttpFault() {
+        return httpFault;
+    }
+
+    public void setHttpFault(XdsHttpFault httpFault) {
+        this.httpFault = httpFault;
+    }
+
+    public XdsRateLimit getRateLimit() {
+        return rateLimit;
+    }
+
+    public void setRateLimit(XdsRateLimit rateLimit) {
+        this.rateLimit = rateLimit;
+    }
 }

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsRouteAction.java

@@ -31,6 +31,8 @@ public class XdsRouteAction {
 
     private XdsWeightedClusters weightedClusters;
 
+    private XdsRetryPolicy retryPolicy;
+
     public String getCluster() {
         return cluster;
     }
@@ -55,6 +57,14 @@ public void setWeightedClusters(XdsWeightedClusters weightedClusters) {
         this.weightedClusters = weightedClusters;
     }
 
+    public XdsRetryPolicy getRetryPolicy() {
+        return retryPolicy;
+    }
+
+    public void setRetryPolicy(XdsRetryPolicy retryPolicy) {
+        this.retryPolicy = retryPolicy;
+    }
+
     /**
      * xDS WeightedClusters
      *

sermant-agentcore/sermant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsServiceCluster.java

@@ -18,6 +18,7 @@
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 
 /**
@@ -104,4 +105,38 @@ public XdsLbPolicy getBaseLbPolicyOfService() {
         }
         return xdsCluster.getLbPolicy();
     }
+
+    /**
+     * get XdsRequestCircuitBreakers
+     *
+     * @param clusterName cluster name
+     * @return XdsOutlierDetection
+     */
+    public Optional<XdsRequestCircuitBreakers> getRequestCircuitBreakersOfCluster(String clusterName) {
+        if (clusters == null) {
+            return Optional.empty();
+        }
+        XdsCluster xdsCluster = clusters.get(clusterName);
+        if (xdsCluster == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(xdsCluster.getRequestCircuitBreakers());
+    }
+
+    /**
+     * get XdsInstanceCircuitBreakers
+     *
+     * @param clusterName cluster name
+     * @return XdsOutlierDetection
+     */
+    public Optional<XdsInstanceCircuitBreakers> getInstanceCircuitBreakersOfCluster(String clusterName) {
+        if (clusters == null) {
+            return Optional.empty();
+        }
+        XdsCluster xdsCluster = clusters.get(clusterName);
+        if (xdsCluster == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(xdsCluster.getInstanceCircuitBreakers());
+    }
 }

sermant-agentcore/sermant-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsAbortTest.java

@@ -29,9 +29,9 @@ public class XdsAbortTest {
     @Test
     public void testXdsAbort() {
         XdsAbort abort = new XdsAbort();
-        abort.setPercentage(0.1f);
+        abort.setPercentage(100);
         abort.setHttpStatus(200);
         Assert.assertEquals(200, abort.getHttpStatus());
-        Assert.assertEquals(0.1f, abort.getPercentage(), 0);
+        Assert.assertEquals(100, abort.getPercentage(), 0);
     }
 }

sermant-agentcore/sermant-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsClusterTest.java

@@ -33,9 +33,15 @@ public void testXdsCluster() {
         cluster.setLbPolicy(XdsLbPolicy.RANDOM);
         cluster.setLocalityLb(true);
         cluster.setServiceName("serviceA");
+        XdsInstanceCircuitBreakers instanceCircuitBreakers = new XdsInstanceCircuitBreakers();
+        cluster.setInstanceCircuitBreakers(instanceCircuitBreakers);
+        XdsRequestCircuitBreakers requestCircuitBreakers = new XdsRequestCircuitBreakers();
+        cluster.setRequestCircuitBreakers(requestCircuitBreakers);
         Assert.assertTrue(cluster.isLocalityLb());
         Assert.assertEquals("outbound|8080||serviceA.default.svc.cluster.local", cluster.getClusterName());
         Assert.assertEquals("serviceA", cluster.getServiceName());
         Assert.assertEquals(XdsLbPolicy.RANDOM, cluster.getLbPolicy());
+        Assert.assertEquals(instanceCircuitBreakers, cluster.getInstanceCircuitBreakers());
+        Assert.assertEquals(requestCircuitBreakers, cluster.getRequestCircuitBreakers());
     }
 }

sermant-agentcore/sermant-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsDelayTest.java

@@ -29,9 +29,9 @@ public class XdsDelayTest {
     @Test
     public void testXdsDelay() {
         XdsDelay delay = new XdsDelay();
-        delay.setPercentage(0.1f);
+        delay.setPercentage(100);
         delay.setFixedDelay(200L);
         Assert.assertEquals(200, delay.getFixedDelay());
-        Assert.assertEquals(0.1f, delay.getPercentage(), 0);
+        Assert.assertEquals(100, delay.getPercentage(), 0);
     }
 }

sermant-agentcore/sermant-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsInstanceCircuitBreakerTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2024-2024 Sermant Authors. All rights reserved.
+ *
+ *   Licensed 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 io.sermant.core.service.xds.entity;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * XdsInstanceCircuitBreakerTest
+ *
+ * @author zhp
+ * @since 2024-11-21
+ **/
+public class XdsInstanceCircuitBreakerTest {
+    @Test
+    public void testXdsOutlierDetection() {
+        XdsInstanceCircuitBreakers xdsInstanceCircuitBreakers = initInstanceCircuitBreakers();
+        Assert.assertEquals(1.0f, xdsInstanceCircuitBreakers.getFailurePercentageMinimumHosts(), 0.0f);
+        Assert.assertEquals(10, xdsInstanceCircuitBreakers.getConsecutiveGatewayFailure(), 0.0f);
+        Assert.assertEquals(10, xdsInstanceCircuitBreakers.getMaxEjectionPercent(), 0.0f);
+        Assert.assertEquals(20, xdsInstanceCircuitBreakers.getConsecutiveLocalOriginFailure(), 0.0f);
+        Assert.assertTrue(xdsInstanceCircuitBreakers.isSplitExternalLocalOriginErrors());
+        Assert.assertEquals(1000L, xdsInstanceCircuitBreakers.getInterval());
+        Assert.assertEquals(1000L, xdsInstanceCircuitBreakers.getBaseEjectionTime());
+        Assert.assertEquals(30, xdsInstanceCircuitBreakers.getConsecutive5xxFailure());
+    }
+
+    private XdsInstanceCircuitBreakers initInstanceCircuitBreakers() {
+        XdsInstanceCircuitBreakers xdsInstanceCircuitBreakers = new XdsInstanceCircuitBreakers();
+        xdsInstanceCircuitBreakers.setFailurePercentageMinimumHosts(1.0f);
+        xdsInstanceCircuitBreakers.setConsecutiveGatewayFailure(10);
+        xdsInstanceCircuitBreakers.setMaxEjectionPercent(10f);
+        xdsInstanceCircuitBreakers.setConsecutiveLocalOriginFailure(20);
+        xdsInstanceCircuitBreakers.setSplitExternalLocalOriginErrors(true);
+        xdsInstanceCircuitBreakers.setInterval(1000L);
+        xdsInstanceCircuitBreakers.setBaseEjectionTime(1000L);
+        xdsInstanceCircuitBreakers.setConsecutive5xxFailure(30);
+        return xdsInstanceCircuitBreakers;
+    }
+}