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.XdsCircuitBreakers;
+import io.sermant.core.service.xds.entity.XdsHttpFault;
+import io.sermant.core.service.xds.entity.XdsOutlierDetection;
+import io.sermant.core.service.xds.entity.XdsRateLimit;
+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 circuit breaker information of cluster
+     *
+     * @param serviceName service name
+     * @param clusterName cluster name
+     * @return circuit breaker rules
+     */
+    Optional<XdsCircuitBreakers> getCircuitBreakers(String serviceName, String clusterName);
+
+    /**
+     * get Outlier Detection of cluster
+     *
+     * @param serviceName service name
+     * @param clusterName cluster name
+     * @return Outlier Detection rules
+     */
+    Optional<XdsOutlierDetection> getOutlierDetection(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/XdsCluster.java

@@ -31,6 +31,10 @@ public class XdsCluster {
 
     private boolean isLocalityLb;
 
+    private XdsOutlierDetection outlierDetection;
+
+    private XdsCircuitBreakers circuitBreakers;
+
     public String getClusterName() {
         return clusterName;
     }
@@ -62,4 +66,20 @@ public boolean isLocalityLb() {
     public void setLocalityLb(boolean localityLb) {
         isLocalityLb = localityLb;
     }
+
+    public XdsOutlierDetection getOutlierDetection() {
+        return outlierDetection;
+    }
+
+    public void setOutlierDetection(XdsOutlierDetection outlierDetection) {
+        this.outlierDetection = outlierDetection;
+    }
+
+    public XdsCircuitBreakers getCircuitBreakers() {
+        return circuitBreakers;
+    }
+
+    public void setCircuitBreakers(XdsCircuitBreakers circuitBreakers) {
+        this.circuitBreakers = circuitBreakers;
+    }
 }

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 Outlier Detection
+     *
+     * @param clusterName cluster name
+     * @return XdsOutlierDetection
+     */
+    public Optional<XdsOutlierDetection> getOutlierDetectionOfCluster(String clusterName) {
+        if (clusters == null) {
+            return Optional.empty();
+        }
+        XdsCluster xdsCluster = clusters.get(clusterName);
+        if (xdsCluster == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(xdsCluster.getOutlierDetection());
+    }
+
+    /**
+     * get CircuitBreakers
+     *
+     * @param clusterName cluster name
+     * @return XdsOutlierDetection
+     */
+    public Optional<XdsCircuitBreakers> getCircuitBreakersOfCluster(String clusterName) {
+        if (clusters == null) {
+            return Optional.empty();
+        }
+        XdsCluster xdsCluster = clusters.get(clusterName);
+        if (xdsCluster == null) {
+            return Optional.empty();
+        }
+        return Optional.ofNullable(xdsCluster.getCircuitBreakers());
+    }
 }

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");
+        XdsCircuitBreakers circuitBreakers = new XdsCircuitBreakers();
+        cluster.setCircuitBreakers(circuitBreakers);
+        XdsOutlierDetection outlierDetection = new XdsOutlierDetection();
+        cluster.setOutlierDetection(outlierDetection);
         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(outlierDetection, cluster.getOutlierDetection());
+        Assert.assertEquals(circuitBreakers, cluster.getCircuitBreakers());
     }
 }

sermant-agentcore/sermant-agentcore-implement/src/main/java/io/sermant/implement/service/xds/XdsCoreServiceImpl.java

@@ -18,12 +18,14 @@
 
 import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.service.xds.XdsCoreService;
+import io.sermant.core.service.xds.XdsFlowControlService;
 import io.sermant.core.service.xds.XdsLoadBalanceService;
 import io.sermant.core.service.xds.XdsRouteService;
 import io.sermant.core.service.xds.XdsServiceDiscovery;
 import io.sermant.implement.service.xds.client.XdsClient;
 import io.sermant.implement.service.xds.discovery.XdsServiceDiscoveryImpl;
 import io.sermant.implement.service.xds.env.XdsConstant;
+import io.sermant.implement.service.xds.flowcontrol.XdsFlowControlServiceImpl;
 import io.sermant.implement.service.xds.handler.CdsHandler;
 import io.sermant.implement.service.xds.handler.EdsHandler;
 import io.sermant.implement.service.xds.handler.LdsHandler;
@@ -50,6 +52,8 @@ public class XdsCoreServiceImpl implements XdsCoreService {
 
     private XdsLoadBalanceService xdsLoadBalanceService;
 
+    private XdsFlowControlService xdsFlowControlService;
+
     private XdsClient client;
 
     @Override
@@ -71,6 +75,7 @@ public void start() {
         xdsServiceDiscovery = new XdsServiceDiscoveryImpl(edsHandler);
         xdsRouteService = new XdsRouteServiceImpl();
         xdsLoadBalanceService = new XdsLoadBalanceServiceImpl();
+        xdsFlowControlService = new XdsFlowControlServiceImpl();
     }
 
     @Override
@@ -96,4 +101,9 @@ public XdsRouteService getXdsRouteService() {
     public XdsLoadBalanceService getLoadBalanceService() {
         return xdsLoadBalanceService;
     }
+
+    @Override
+    public XdsFlowControlService getXdsFlowControlService() {
+        return xdsFlowControlService;
+    }
 }

sermant-agentcore/sermant-agentcore-implement/src/main/java/io/sermant/implement/service/xds/flowcontrol/XdsFlowControlServiceImpl.java

@@ -0,0 +1,101 @@
+/*
+ * 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.implement.service.xds.flowcontrol;
+
+import io.sermant.core.service.xds.XdsFlowControlService;
+import io.sermant.core.service.xds.entity.XdsCircuitBreakers;
+import io.sermant.core.service.xds.entity.XdsHttpFault;
+import io.sermant.core.service.xds.entity.XdsOutlierDetection;
+import io.sermant.core.service.xds.entity.XdsRateLimit;
+import io.sermant.core.service.xds.entity.XdsRetryPolicy;
+import io.sermant.core.service.xds.entity.XdsRoute;
+import io.sermant.core.service.xds.entity.XdsRouteAction;
+import io.sermant.core.service.xds.entity.XdsServiceCluster;
+import io.sermant.core.utils.StringUtils;
+import io.sermant.implement.service.xds.cache.XdsDataCache;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * The implementation class of XdsFlowControlService
+ *
+ * @author zhp
+ * @since 2024-11-27
+ **/
+public class XdsFlowControlServiceImpl implements XdsFlowControlService {
+    /**
+     * constructor
+     */
+    public XdsFlowControlServiceImpl() {
+    }
+
+    @Override
+    public Optional<XdsCircuitBreakers> getCircuitBreakers(String serviceName, String clusterName) {
+        Map<String, XdsServiceCluster> serviceClusterMap = XdsDataCache.getServiceClusterMap();
+        XdsServiceCluster serviceCluster = serviceClusterMap.get(serviceName);
+        if (serviceCluster == null) {
+            return Optional.empty();
+        }
+        return serviceCluster.getCircuitBreakersOfCluster(clusterName);
+    }
+
+    @Override
+    public Optional<XdsOutlierDetection> getOutlierDetection(String serviceName, String clusterName) {
+        Map<String, XdsServiceCluster> serviceClusterMap = XdsDataCache.getServiceClusterMap();
+        XdsServiceCluster serviceCluster = serviceClusterMap.get(serviceName);
+        if (serviceCluster == null) {
+            return Optional.empty();
+        }
+        return serviceCluster.getOutlierDetectionOfCluster(clusterName);
+    }
+
+    @Override
+    public Optional<XdsRetryPolicy> getRetryPolicy(String serviceName, String routeName) {
+        List<XdsRoute> xdsRoutes = XdsDataCache.getServiceRoute(serviceName);
+        for (XdsRoute xdsRoute : xdsRoutes) {
+            if (StringUtils.equals(xdsRoute.getName(), routeName)) {
+                XdsRouteAction routeAction = xdsRoute.getRouteAction();
+                return routeAction == null ? Optional.empty() : Optional.ofNullable(routeAction.getRetryPolicy());
+            }
+        }
+        return Optional.empty();
+    }
+
+    @Override
+    public Optional<XdsRateLimit> getRateLimit(String serviceName, String routeName) {
+        List<XdsRoute> xdsRoutes = XdsDataCache.getServiceRoute(serviceName);
+        for (XdsRoute xdsRoute : xdsRoutes) {
+            if (StringUtils.equals(xdsRoute.getName(), routeName)) {
+                return Optional.ofNullable(xdsRoute.getRateLimit());
+            }
+        }
+        return Optional.empty();
+    }
+
+    @Override
+    public Optional<XdsHttpFault> getHttpFault(String serviceName, String routeName) {
+        List<XdsRoute> xdsRoutes = XdsDataCache.getServiceRoute(serviceName);
+        for (XdsRoute xdsRoute : xdsRoutes) {
+            if (StringUtils.equals(xdsRoute.getName(), routeName)) {
+                return Optional.ofNullable(xdsRoute.getHttpFault());
+            }
+        }
+        return Optional.empty();
+    }
+}