fixed failover with round robin load balancing (DE-724)

rashtao · rashtao · commit 6740948e5894 · 2023-11-30T21:54:19.000+01:00
diff --git a/.github/workflows/resilience.yml b/.github/workflows/resilience.yml
@@ -23,7 +23,7 @@ jobs:
     runs-on: ubuntu-latest
 
     env:
-      TOXIPROXY_VERSION: v2.5.0
+      TOXIPROXY_VERSION: v2.7.0
 
     strategy:
       fail-fast: false
diff --git a/core/src/main/java/com/arangodb/internal/net/DirtyReadHostHandler.java b/core/src/main/java/com/arangodb/internal/net/DirtyReadHostHandler.java
@@ -52,6 +52,12 @@ public Host get(final HostHandle hostHandle, final AccessType accessType) {
         return determineHostHandler().get(hostHandle, accessType);
     }
 
+    @Override
+    public boolean hasNext(HostHandle hostHandle, AccessType accessType) {
+        this.currentAccessType = accessType;
+        return determineHostHandler().hasNext(hostHandle, accessType);
+    }
+
     @Override
     public void success() {
         determineHostHandler().success();
diff --git a/core/src/main/java/com/arangodb/internal/net/FallbackHostHandler.java b/core/src/main/java/com/arangodb/internal/net/FallbackHostHandler.java
@@ -49,7 +49,7 @@ public FallbackHostHandler(final HostResolver resolver) {
 
     @Override
     public Host get(final HostHandle hostHandle, AccessType accessType) {
-        if (current != lastSuccess || iterations < 3) {
+        if (hasNext(hostHandle, accessType)) {
             return current;
         } else {
             ArangoDBException e = ArangoDBException.of("Cannot contact any host!",
@@ -59,6 +59,11 @@ public Host get(final HostHandle hostHandle, AccessType accessType) {
         }
     }
 
+    @Override
+    public boolean hasNext(HostHandle hostHandle, AccessType accessType) {
+        return current != lastSuccess || iterations < 3;
+    }
+
     @Override
     public void success() {
         lastSuccess = current;
diff --git a/core/src/main/java/com/arangodb/internal/net/HostHandler.java b/core/src/main/java/com/arangodb/internal/net/HostHandler.java
@@ -31,6 +31,8 @@ public interface HostHandler {
 
     Host get(HostHandle hostHandle, AccessType accessType);
 
+    boolean hasNext(HostHandle hostHandle, AccessType accessType);
+
     void success();
 
     void fail(Exception exception);
diff --git a/core/src/main/java/com/arangodb/internal/net/RandomHostHandler.java b/core/src/main/java/com/arangodb/internal/net/RandomHostHandler.java
@@ -52,6 +52,11 @@ public Host get(final HostHandle hostHandle, AccessType accessType) {
         return current;
     }
 
+    @Override
+    public boolean hasNext(HostHandle hostHandle, AccessType accessType) {
+        return true;
+    }
+
     @Override
     public void success() {
         fallback.success();
diff --git a/core/src/main/java/com/arangodb/internal/net/RoundRobinHostHandler.java b/core/src/main/java/com/arangodb/internal/net/RoundRobinHostHandler.java
@@ -23,6 +23,8 @@
 import com.arangodb.ArangoDBException;
 import com.arangodb.ArangoDBMultipleException;
 import com.arangodb.config.HostDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -32,6 +34,8 @@
  */
 public class RoundRobinHostHandler implements HostHandler {
 
+    private final static Logger LOGGER = LoggerFactory.getLogger(RoundRobinHostHandler.class);
+
     private final HostResolver resolver;
     private final List<Exception> lastFailExceptions;
     private long current;
@@ -74,9 +78,17 @@ public Host get(final HostHandle hostHandle, AccessType accessType) {
                 hostHandle.setHost(host.getDescription());
             }
         }
+        LOGGER.debug("Returning host: {}", host);
         return host;
     }
 
+    @Override
+    public boolean hasNext(HostHandle hostHandle, AccessType accessType) {
+        hosts = resolver.getHosts();
+        int size = hosts.getHostsList().size();
+        return  fails <= size;
+    }
+
     @Override
     public void success() {
         reset();
diff --git a/http/src/main/java/com/arangodb/http/HttpCommunication.java b/http/src/main/java/com/arangodb/http/HttpCommunication.java
@@ -121,7 +121,7 @@ private CompletableFuture<InternalResponse> executeAsync(final InternalRequest r
                             }
                         }
                     } catch (Exception ex) {
-                        rfuture.completeExceptionally(ArangoDBException.of(ex));
+                        rfuture.completeExceptionally(ArangoDBException.of(ex, reqId));
                     }
                 });
         return rfuture;
@@ -134,8 +134,9 @@ private void handleException(boolean isSafe, Throwable e, HostHandle hostHandle,
         if (hostHandle != null && hostHandle.getHost() != null) {
             hostHandle.setHost(null);
         }
-        Host nextHost = hostHandler.get(hostHandle, RequestUtils.determineAccessType(request));
-        if (nextHost != null && isSafe) {
+        boolean hasNextHost = hostHandler.hasNext(hostHandle, RequestUtils.determineAccessType(request));
+        if (hasNextHost && isSafe) {
+            Host nextHost = hostHandler.get(hostHandle, RequestUtils.determineAccessType(request));
             LOGGER.warn("Could not connect to {} while executing request [id={}]",
                     host.getDescription(), reqId, ioEx);
             LOGGER.debug("Try connecting to {}", nextHost.getDescription());
@@ -145,7 +146,6 @@ private void handleException(boolean isSafe, Throwable e, HostHandle hostHandle,
             );
         } else {
             ArangoDBException aEx = ArangoDBException.of(ioEx, reqId);
-            LOGGER.error(aEx.getMessage(), aEx);
             rfuture.completeExceptionally(aEx);
         }
     }
diff --git a/resilience-tests/src/test/java/resilience/ClusterTest.java b/resilience-tests/src/test/java/resilience/ClusterTest.java
@@ -4,13 +4,13 @@
 import com.arangodb.ArangoDBAsync;
 import com.arangodb.Request;
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import resilience.utils.MemoryAppender;
 import eu.rekawek.toxiproxy.Proxy;
 import eu.rekawek.toxiproxy.ToxiproxyClient;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
+import resilience.utils.MemoryAppender;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -117,7 +117,14 @@ protected static String serverIdPOST(ArangoDBAsync adb) {
                     .get("serverInfo")
                     .get("serverId")
                     .textValue();
-        } catch (InterruptedException | ExecutionException e) {
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof RuntimeException) {
+                throw (RuntimeException) cause;
+            } else {
+                throw new RuntimeException(e);
+            }
+        } catch (InterruptedException e) {
             throw new RuntimeException(e);
         }
     }
diff --git a/resilience-tests/src/test/java/resilience/loadbalance/LoadBalanceRoundRobinClusterTest.java b/resilience-tests/src/test/java/resilience/loadbalance/LoadBalanceRoundRobinClusterTest.java
@@ -79,7 +79,7 @@ void failoverAsync(ArangoDBAsync arangoDB) throws IOException {
         assertThat(serverIdGET(arangoDB)).isEqualTo(endpoints.get(0).getServerId());
     }
 
-    // FIXME: this only passes for HTTP, while fails for VST and HTTP2
+    // FIXME: this fails for VST
     @Disabled
     @ParameterizedTest(name = "{index}")
     @MethodSource("arangoProvider")
@@ -105,13 +105,39 @@ void retryGET(ArangoDB arangoDB) throws IOException, InterruptedException {
         es.shutdown();
     }
 
-    // FIXME: this only passes for VST, while fails for HTTP and HTTP2:
-    // for HTTP and HTTP2 it skips 2 hosts instead of skipping only the failing one
+    // FIXME: this fails for VST
     @Disabled
+    @ParameterizedTest(name = "{index}")
+    @MethodSource("asyncArangoProvider")
+    void retryGETAsync(ArangoDBAsync arangoDB) throws IOException, InterruptedException {
+        List<Endpoint> endpoints = getEndpoints();
+
+        // slow down the driver connection
+        Latency toxic = getEndpoints().get(0).getProxy().toxics().latency("latency", ToxicDirection.DOWNSTREAM, 10_000);
+        Thread.sleep(100);
+
+        ScheduledExecutorService es = Executors.newSingleThreadScheduledExecutor();
+        es.schedule(() -> getEndpoints().get(0).disable(), 300, TimeUnit.MILLISECONDS);
+
+        assertThat(serverIdGET(arangoDB)).isEqualTo(endpoints.get(1).getServerId());
+        assertThat(serverIdGET(arangoDB)).isEqualTo(endpoints.get(2).getServerId());
+
+        toxic.remove();
+        enableAllEndpoints();
+        Thread.sleep(100);
+
+        assertThat(serverIdGET(arangoDB)).isEqualTo(endpoints.get(0).getServerId());
+
+        es.shutdown();
+    }
+
     @ParameterizedTest(name = "{index}")
     @MethodSource("arangoProvider")
     void retryPOST(ArangoDB arangoDB) throws IOException, InterruptedException {
         List<Endpoint> endpoints = getEndpoints();
+        for (Endpoint endpoint : endpoints) {
+            System.out.println(endpoint.getServerId());
+        }
 
         // slow down the driver connection
         Latency toxic = getEndpoints().get(0).getProxy().toxics().latency("latency", ToxicDirection.DOWNSTREAM, 10_000);
@@ -120,7 +146,7 @@ void retryPOST(ArangoDB arangoDB) throws IOException, InterruptedException {
         ScheduledExecutorService es = Executors.newSingleThreadScheduledExecutor();
         es.schedule(() -> getEndpoints().get(0).disable(), 300, TimeUnit.MILLISECONDS);
 
-        Throwable thrown = catchThrowable(()->serverIdPOST(arangoDB));
+        Throwable thrown = catchThrowable(() -> serverIdPOST(arangoDB));
         assertThat(thrown).isInstanceOf(ArangoDBException.class);
         assertThat(thrown.getCause()).isInstanceOf(IOException.class);
 
@@ -129,8 +155,37 @@ void retryPOST(ArangoDB arangoDB) throws IOException, InterruptedException {
 
         toxic.remove();
         enableAllEndpoints();
+
+        assertThat(serverIdPOST(arangoDB)).isEqualTo(endpoints.get(0).getServerId());
+
+        es.shutdown();
+    }
+
+    @ParameterizedTest(name = "{index}")
+    @MethodSource("asyncArangoProvider")
+    void retryPOSTAsync(ArangoDBAsync arangoDB) throws IOException, InterruptedException {
+        List<Endpoint> endpoints = getEndpoints();
+        for (Endpoint endpoint : endpoints) {
+            System.out.println(endpoint.getServerId());
+        }
+
+        // slow down the driver connection
+        Latency toxic = getEndpoints().get(0).getProxy().toxics().latency("latency", ToxicDirection.DOWNSTREAM, 10_000);
         Thread.sleep(100);
 
+        ScheduledExecutorService es = Executors.newSingleThreadScheduledExecutor();
+        es.schedule(() -> getEndpoints().get(0).disable(), 300, TimeUnit.MILLISECONDS);
+
+        Throwable thrown = catchThrowable(() -> serverIdPOST(arangoDB));
+        assertThat(thrown).isInstanceOf(ArangoDBException.class);
+        assertThat(thrown.getCause()).isInstanceOf(IOException.class);
+
+        assertThat(serverIdPOST(arangoDB)).isEqualTo(endpoints.get(1).getServerId());
+        assertThat(serverIdPOST(arangoDB)).isEqualTo(endpoints.get(2).getServerId());
+
+        toxic.remove();
+        enableAllEndpoints();
+
         assertThat(serverIdPOST(arangoDB)).isEqualTo(endpoints.get(0).getServerId());
 
         es.shutdown();

Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,7 @@ private CompletableFuture<InternalResponse> executeAsync(final InternalRequest r`
`121`	`121`	`}`
`122`	`122`	`}`
`123`	`123`	`} catch (Exception ex) {`
`124`		`- rfuture.completeExceptionally(ArangoDBException.of(ex));`
	`124`	`+ rfuture.completeExceptionally(ArangoDBException.of(ex, reqId));`
`125`	`125`	`}`
`126`	`126`	`});`
`127`	`127`	`return rfuture;`
`@@ -134,8 +134,9 @@ private void handleException(boolean isSafe, Throwable e, HostHandle hostHandle,`
`134`	`134`	`if (hostHandle != null && hostHandle.getHost() != null) {`
`135`	`135`	`hostHandle.setHost(null);`
`136`	`136`	`}`
`137`		`- Host nextHost = hostHandler.get(hostHandle, RequestUtils.determineAccessType(request));`
`138`		`- if (nextHost != null && isSafe) {`
	`137`	`+ boolean hasNextHost = hostHandler.hasNext(hostHandle, RequestUtils.determineAccessType(request));`
	`138`	`+ if (hasNextHost && isSafe) {`
	`139`	`+ Host nextHost = hostHandler.get(hostHandle, RequestUtils.determineAccessType(request));`
`139`	`140`	`LOGGER.warn("Could not connect to {} while executing request [id={}]",`
`140`	`141`	`host.getDescription(), reqId, ioEx);`
`141`	`142`	`LOGGER.debug("Try connecting to {}", nextHost.getDescription());`
`@@ -145,7 +146,6 @@ private void handleException(boolean isSafe, Throwable e, HostHandle hostHandle,`
`145`	`146`	`);`
`146`	`147`	`} else {`
`147`	`148`	`ArangoDBException aEx = ArangoDBException.of(ioEx, reqId);`
`148`		`- LOGGER.error(aEx.getMessage(), aEx);`
`149`	`149`	`rfuture.completeExceptionally(aEx);`
`150`	`150`	`}`
`151`	`151`	`}`