Fix some bug in xds flow control

Signed-off-by: hanbingleixue <[email protected]>
sermant-io · Jan 3, 2025 · 7b0fd62 · 7b0fd62
1 parent a5de30d
commit 7b0fd62
Show file tree

Hide file tree

Showing 29 changed files with 316 additions and 124 deletions.
diff --git a/...rmant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsRetryPolicy.java b/...rmant-agentcore-core/src/main/java/io/sermant/core/service/xds/entity/XdsRetryPolicy.java
@@ -16,6 +16,8 @@
 
 package io.sermant.core.service.xds.entity;
 
+import java.util.List;
+
 /**
  * Xds Retry Policy information
  *
@@ -37,12 +39,7 @@ public class XdsRetryPolicy {
     /**
      * Specifies the conditions under which retry takes place
      */
-    private String retryOn;
-
-    /**
-     * Specifies the conditions under which retry takes place
-     */
-    private String retryHostPredicate;
+    private List<String> retryConditions;
 
     public long getMaxAttempts() {
         return maxAttempts;
@@ -60,19 +57,11 @@ public void setPerTryTimeout(long perTryTimeout) {
         this.perTryTimeout = perTryTimeout;
     }
 
-    public String getRetryOn() {
-        return retryOn;
-    }
-
-    public void setRetryOn(String retryOn) {
-        this.retryOn = retryOn;
-    }
-
-    public String getRetryHostPredicate() {
-        return retryHostPredicate;
+    public List<String> getRetryConditions() {
+        return retryConditions;
     }
 
-    public void setRetryHostPredicate(String retryHostPredicate) {
-        this.retryHostPredicate = retryHostPredicate;
+    public void setRetryConditions(List<String> retryConditions) {
+        this.retryConditions = retryConditions;
     }
 }
diff --git a/...t-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsRetryPolicyTest.java b/...t-agentcore-core/src/test/java/io/sermant/core/service/xds/entity/XdsRetryPolicyTest.java
@@ -19,6 +19,8 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.Collections;
+
 /**
  * XdsRetryPolicyTest
  *
@@ -29,12 +31,10 @@ public class XdsRetryPolicyTest {
     @Test
     public void testXdsRetryPolicy() {
         XdsRetryPolicy xdsRetryPolicy = new XdsRetryPolicy();
-        xdsRetryPolicy.setRetryHostPredicate("PreviousHostsPredicate");
-        xdsRetryPolicy.setRetryOn("503");
+        xdsRetryPolicy.setRetryConditions(Collections.singletonList("503"));
         xdsRetryPolicy.setMaxAttempts(8);
         xdsRetryPolicy.setPerTryTimeout(2000);
-        Assert.assertEquals("PreviousHostsPredicate", xdsRetryPolicy.getRetryHostPredicate());
-        Assert.assertEquals("503", xdsRetryPolicy.getRetryOn());
+        Assert.assertEquals("503", xdsRetryPolicy.getRetryConditions().get(0));
         Assert.assertEquals(8, xdsRetryPolicy.getMaxAttempts());
         Assert.assertEquals(2000, xdsRetryPolicy.getPerTryTimeout());
     }

diff --git a/...mplement/src/main/java/io/sermant/implement/service/xds/utils/RdsProtocolTransformer.java b/...mplement/src/main/java/io/sermant/implement/service/xds/utils/RdsProtocolTransformer.java
@@ -38,6 +38,7 @@
 import io.envoyproxy.envoy.extensions.filters.http.fault.v3.HTTPFault;
 import io.envoyproxy.envoy.type.matcher.v3.StringMatcher;
 import io.envoyproxy.envoy.type.v3.FractionalPercent;
+import io.sermant.core.common.CommonConstant;
 import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.service.xds.entity.XdsAbort;
 import io.sermant.core.service.xds.entity.XdsDelay;
@@ -68,6 +69,7 @@
 
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Locale;
@@ -264,13 +266,12 @@ private static XdsClusterWeight parseClusterWeight(ClusterWeight clusterWeight)
 
     private static XdsRetryPolicy parseRetryPolicy(RetryPolicy retryPolicy) {
         XdsRetryPolicy xdsRetryPolicy = new XdsRetryPolicy();
-        xdsRetryPolicy.setRetryOn(retryPolicy.getRetryOn());
+        if (!StringUtils.isEmpty(retryPolicy.getRetryOn())) {
+            xdsRetryPolicy.setRetryConditions(Arrays.asList(retryPolicy.getRetryOn().split(CommonConstant.COMMA)));
+        }
         xdsRetryPolicy.setMaxAttempts(retryPolicy.getHostSelectionRetryMaxAttempts());
         long perTryTimeout = Duration.ofSeconds(retryPolicy.getPerTryTimeout().getSeconds()).toMillis();
         xdsRetryPolicy.setPerTryTimeout(perTryTimeout);
-        if (retryPolicy.getRetryHostPredicateCount() != 0) {
-            xdsRetryPolicy.setRetryHostPredicate(retryPolicy.getRetryHostPredicate(0).getName());
-        }
         return xdsRetryPolicy;
     }
 

diff --git a/...ntrol-common/src/main/java/io/sermant/flowcontrol/common/handler/retry/AbstractRetry.java b/...ntrol-common/src/main/java/io/sermant/flowcontrol/common/handler/retry/AbstractRetry.java
@@ -19,8 +19,12 @@
 
 import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.plugin.config.PluginConfigManager;
+import io.sermant.core.service.xds.entity.XdsRetryPolicy;
+import io.sermant.core.utils.CollectionUtils;
 import io.sermant.flowcontrol.common.config.FlowControlConfig;
 import io.sermant.flowcontrol.common.support.ReflectMethodCacheSupport;
+import io.sermant.flowcontrol.common.xds.retry.RetryCondition;
+import io.sermant.flowcontrol.common.xds.retry.RetryConditionType;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -63,16 +67,62 @@ protected final Class<? extends Throwable>[] findClass(String[] classNames) {
     }
 
     @Override
-    public boolean needRetry(Set<String> statusList, Object result) {
+    public boolean isNeedRetry(Set<String> statusList, Object result) {
         if (result == null) {
             return false;
         }
         final Optional<String> code = getCode(result);
         return code.filter(statusList::contains).isPresent();
     }
 
+    @Override
+    public boolean isNeedRetry(Object result, XdsRetryPolicy retryPolicy) {
+        if (result == null) {
+            return false;
+        }
+        List<String> conditions = retryPolicy.getRetryConditions();
+        if (CollectionUtils.isEmpty(conditions)) {
+            return false;
+        }
+        Optional<String> statusCodeOptional = this.getCode(result);
+        if (!statusCodeOptional.isPresent()) {
+            return false;
+        }
+        String statusCode = statusCodeOptional.get();
+        if (conditions.contains(statusCode)) {
+            return true;
+        }
+        for (String conditionName : conditions) {
+            Optional<RetryCondition> retryConditionOptional = RetryConditionType.getRetryConditionByName(conditionName);
+            if (!retryConditionOptional.isPresent()) {
+                continue;
+            }
+            if (retryConditionOptional.get().needRetry(this, null, statusCode, result)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public boolean isNeedRetry(Throwable ex, XdsRetryPolicy retryPolicy) {
+        if (ex == null) {
+            return false;
+        }
+        for (String conditionName : retryPolicy.getRetryConditions()) {
+            Optional<RetryCondition> retryConditionOptional = RetryConditionType.getRetryConditionByName(conditionName);
+            if (!retryConditionOptional.isPresent()) {
+                continue;
+            }
+            if (retryConditionOptional.get().needRetry(null, ex, null, null)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     /**
-     * implemented by subclasses， if subclass implement {@link #needRetry(Set, Object)}, no need to implement the get
+     * implemented by subclasses， if subclass implement {@link #isNeedRetry(Set, Object)}, no need to implement the get
      * code method
      *
      * @param result interface response result

diff --git a/...l/flowcontrol-common/src/main/java/io/sermant/flowcontrol/common/handler/retry/Retry.java b/...l/flowcontrol-common/src/main/java/io/sermant/flowcontrol/common/handler/retry/Retry.java
@@ -17,6 +17,8 @@
 
 package io.sermant.flowcontrol.common.handler.retry;
 
+import io.sermant.core.service.xds.entity.XdsRetryPolicy;
+
 import java.util.Optional;
 import java.util.Set;
 
@@ -28,13 +30,33 @@
  */
 public interface Retry {
     /**
-     * needToRetry
+     * Retry based on the request result. Retrying is required if the request status is in the statusList.
      *
      * @param statusList List of status codes, valid only for http applications
      * @param result responseResult
      * @return retryOrNot
      */
-    boolean needRetry(Set<String> statusList, Object result);
+    boolean isNeedRetry(Set<String> statusList, Object result);
+
+    /**
+     * Retry based on the request result. If the request result meets the retry conditions in the retry policy,
+     * a retry will be executed
+     *
+     * @param result responseResult
+     * @param retryPolicy retry policy information
+     * @return retryOrNot
+     */
+    boolean isNeedRetry(Object result, XdsRetryPolicy retryPolicy);
+
+    /**
+     * Retry based on the throwable. If the throwable during the execution of the request method meets the retry
+     * conditions in the retry policy, a retry will be executed
+     *
+     * @param throwable Exception thrown during retry
+     * @param retryPolicy Xds Retry Policy information
+     * @return retryOrNot
+     */
+    boolean isNeedRetry(Throwable throwable, XdsRetryPolicy retryPolicy);
 
     /**
      * define which exceptions need to be retried

diff --git a/...l/flowcontrol-common/src/main/java/io/sermant/flowcontrol/common/util/XdsRouterUtils.java b/...l/flowcontrol-common/src/main/java/io/sermant/flowcontrol/common/util/XdsRouterUtils.java
@@ -55,12 +55,13 @@ private XdsRouterUtils() {
      */
     public static Optional<XdsLocality> getLocalityInfoOfSelfService() {
         if (localityObtainedFlag) {
-            return Optional.of(selfServiceLocality);
+            return Optional.ofNullable(selfServiceLocality);
         }
         synchronized (XdsRouterUtils.class) {
             if (localityObtainedFlag) {
-                return Optional.of(selfServiceLocality);
+                return Optional.ofNullable(selfServiceLocality);
             }
+            localityObtainedFlag = true;
             String podIp = NetworkUtils.getKubernetesPodIp();
             if (StringUtils.isEmpty(podIp)) {
                 return Optional.empty();

diff --git a/...owcontrol-common/src/main/java/io/sermant/flowcontrol/common/util/XdsThreadLocalUtil.java b/...owcontrol-common/src/main/java/io/sermant/flowcontrol/common/util/XdsThreadLocalUtil.java
@@ -18,6 +18,8 @@
 
 import io.sermant.flowcontrol.common.entity.FlowControlScenario;
 
+import java.net.HttpURLConnection;
+
 /**
  * xds thread local utility class
  *
@@ -29,6 +31,10 @@ public class XdsThreadLocalUtil {
 
     private static final ThreadLocal<FlowControlScenario> FLOW_CONTROL_SCENARIO_THREAD_LOCAL = new ThreadLocal<>();
 
+    private static final ThreadLocal<HttpURLConnection> CONNECTION_IN_CURRENT_THREAD_LOCAL = new ThreadLocal<>();
+
+    private static final ThreadLocal<Boolean> IS_CONNECTED = new ThreadLocal<>();
+
     private XdsThreadLocalUtil() {
     }
 
@@ -57,6 +63,31 @@ public static void removeSendByteFlag() {
         SEND_BYTE_FLAG.remove();
     }
 
+    /**
+     * save the instance object of HttpURLConnection
+     *
+     * @param connection the instance object of HttpURLConnection
+     */
+    public static void saveHttpUrlConnection(HttpURLConnection connection) {
+        CONNECTION_IN_CURRENT_THREAD_LOCAL.set(connection);
+    }
+
+    /**
+     * get the instance object of HttpURLConnection
+     *
+     * @return the instance object of HttpURLConnection
+     */
+    public static HttpURLConnection getHttpUrlConnection() {
+        return CONNECTION_IN_CURRENT_THREAD_LOCAL.get();
+    }
+
+    /**
+     * remove the instance object of HttpURLConnection
+     */
+    public static void removeHttpUrlConnection() {
+        CONNECTION_IN_CURRENT_THREAD_LOCAL.remove();
+    }
+
     /**
      * Set scenario information
      *
@@ -81,4 +112,29 @@ public static FlowControlScenario getScenarioInfo() {
     public static void removeScenarioInfo() {
         FLOW_CONTROL_SCENARIO_THREAD_LOCAL.remove();
     }
+
+    /**
+     * Set the connection status of the httpUrlConnection
+     *
+     * @param executeStatus the execution status of the connect method, true: executed, false: Not executed
+     */
+    public static void setConnectionStatus(boolean executeStatus) {
+        IS_CONNECTED.set(executeStatus);
+    }
+
+    /**
+     * Is it already connected
+     *
+     * @return connection status
+     */
+    public static boolean isConnected() {
+        return IS_CONNECTED.get() != null && IS_CONNECTED.get();
+    }
+
+    /**
+     * remove the connection status of the httpUrlConnection
+     */
+    public static void removeConnectionStatus() {
+        IS_CONNECTED.remove();
+    }
 }
diff --git a/...mon/src/main/java/io/sermant/flowcontrol/common/xds/circuit/XdsCircuitBreakerManager.java b/...mon/src/main/java/io/sermant/flowcontrol/common/xds/circuit/XdsCircuitBreakerManager.java
@@ -100,7 +100,7 @@ public static void setCircuitBeakerStatus(XdsInstanceCircuitBreakers circuitBrea
         XdsCircuitBreakerInfo circuitBreakerInfo = getCircuitBreakerInfo(scenarioInfo.getServiceName(),
                 scenarioInfo.getRouteName(), scenarioInfo.getAddress());
         if (!XdsThreadLocalUtil.getSendByteFlag() && circuitBreakers.isSplitExternalLocalOriginErrors()
-                && shouldCircuitBreakerByFailure(circuitBreakerInfo.getGateWayFailure(),
+                && shouldCircuitBreakerByFailure(circuitBreakerInfo.getLocalFailure(),
                 circuitBreakers.getConsecutiveLocalOriginFailure(), circuitBreakers.getInterval())) {
             openCircuitBreaker(circuitBreakerInfo, circuitBreakers.getInterval());
         }

diff --git a/...ol-common/src/test/java/io/sermant/flowcontrol/common/handler/retry/RetryContextTest.java b/...ol-common/src/test/java/io/sermant/flowcontrol/common/handler/retry/RetryContextTest.java
@@ -17,6 +17,7 @@
 
 package io.sermant.flowcontrol.common.handler.retry;
 
+import io.sermant.core.service.xds.entity.XdsRetryPolicy;
 import io.sermant.flowcontrol.common.core.rule.RetryRule;
 
 import org.junit.Assert;
@@ -75,7 +76,17 @@ private void muteRetry(Object instance) {
     private Retry buildRetry() {
         return new Retry() {
             @Override
-            public boolean needRetry(Set<String> statusList, Object result) {
+            public boolean isNeedRetry(Set<String> statusList, Object result) {
+                return false;
+            }
+
+            @Override
+            public boolean isNeedRetry(Object result, XdsRetryPolicy retryPolicy) {
+                return false;
+            }
+
+            @Override
+            public boolean isNeedRetry(Throwable throwable, XdsRetryPolicy retryPolicy) {
                 return false;
             }
 

diff --git a/.../flowcontrol-common/src/test/java/io/sermant/flowcontrol/common/util/XdsAbstractTest.java b/.../flowcontrol-common/src/test/java/io/sermant/flowcontrol/common/util/XdsAbstractTest.java
@@ -124,7 +124,7 @@ public void setUp() {
         Mockito.when(xdsCoreService.getLoadBalanceService()).thenReturn(xdsLoadBalanceService);
         requestCircuitBreakers.setMaxRequests(1000);
         instanceCircuitBreakers.setInterval(1000);
-        retryPolicy.setRetryOn("503");
+        retryPolicy.setRetryConditions(Collections.singletonList("503"));
         rateLimit.setResponseHeaderOption(Collections.singletonList(new XdsHeaderOption()));
         XdsDelay delay = new XdsDelay();
         delay.setFixedDelay(1000);

diff --git a/...ontrol-common/src/test/java/io/sermant/flowcontrol/common/xds/handler/XdsHandlerTest.java b/...ontrol-common/src/test/java/io/sermant/flowcontrol/common/xds/handler/XdsHandlerTest.java
@@ -63,7 +63,7 @@ public void testGetRetryPolicy() {
         final Optional<XdsRetryPolicy> result = XdsHandler.INSTANCE.getRetryPolicy(
                 SERVICE_NAME, ROUTE_NAME);
         Assert.assertTrue(result.isPresent());
-        Assert.assertEquals(retryPolicy.getRetryOn(), result.get().getRetryOn());
+        Assert.assertEquals(retryPolicy.getRetryConditions().get(0), result.get().getRetryConditions().get(0));
     }
 
     @Test