Repository: mageddo/dns-proxy-server (java)

Issue:
Circuit Breaker for Remote DNS Servers
### What is expected

DPS  have a remote DNS circuit breaker feature to automatically disable/re-enable remote servers when the come offline or online again.

### What is Happening

DPS will try the offline remote server every time a query is made and wait until the timeout to go to the next server.

### Changes
* After 3 failures DPS will open the circuit for the failing remote server and only check it again after 20 seconds

Pull request description:
Circuit Breaker for Remote DNS Servers
Fixing #415 

Patch:
diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md
index 33dc742d..d0ec8598 100644
--- a/RELEASE-NOTES.md
+++ b/RELEASE-NOTES.md
@@ -1,3 +1,6 @@
+## 3.16
+* Circuit Breaker for Remote DNS Servers #415
+
 ## 3.15
 * Support for AAAA records on Docker container, LocalDB and System solving.
 * AAAA GUI records support 
diff --git a/build.gradle b/build.gradle
index 7103ef15..4fddafc3 100644
--- a/build.gradle
+++ b/build.gradle
@@ -44,6 +44,7 @@ dependencies {
   implementation('dnsjava:dnsjava:3.5.2')
   implementation('org.apache.commons:commons-compress:1.22')
   implementation('org.rauschig:jarchivelib:1.2.0')
+  implementation('dev.failsafe:failsafe:3.3.1')
 
   implementation('com.github.docker-java:docker-java-core:3.2.14')
   implementation('com.github.docker-java:docker-java-transport-httpclient5:3.2.14')
diff --git a/gradle.properties b/gradle.properties
index 26ab97ed..0b0b9996 100644
--- a/gradle.properties
+++ b/gradle.properties
@@ -1,1 +1,1 @@
-version=3.15.13-snapshot
+version=3.16.0-snapshot
diff --git a/src/main/java/com/mageddo/commons/circuitbreaker/CircuitCheckException.java b/src/main/java/com/mageddo/commons/circuitbreaker/CircuitCheckException.java
new file mode 100644
index 00000000..33ba6839
--- /dev/null
+++ b/src/main/java/com/mageddo/commons/circuitbreaker/CircuitCheckException.java
@@ -0,0 +1,11 @@
+package com.mageddo.commons.circuitbreaker;
+
+public class CircuitCheckException extends RuntimeException {
+  public CircuitCheckException(String message) {
+    super(message);
+  }
+
+  public CircuitCheckException(Throwable e) {
+    super(e.getMessage(), e);
+  }
+}
diff --git a/src/main/java/com/mageddo/dnsproxyserver/quarkus/QuarkusConfig.java b/src/main/java/com/mageddo/dnsproxyserver/quarkus/QuarkusConfig.java
index 2e25a7d2..4b24aa5e 100644
--- a/src/main/java/com/mageddo/dnsproxyserver/quarkus/QuarkusConfig.java
+++ b/src/main/java/com/mageddo/dnsproxyserver/quarkus/QuarkusConfig.java
@@ -1,13 +1,13 @@
 package com.mageddo.dnsproxyserver.quarkus;
 
 import com.mageddo.dnsproxyserver.config.Configs;
+import com.mageddo.dnsproxyserver.server.dns.solver.Resolver;
+import com.mageddo.dnsproxyserver.server.dns.solver.SimpleResolver;
 import com.mageddo.net.IpAddr;
 import com.mageddo.dnsproxyserver.server.dns.solver.RemoteResolvers;
 import com.mageddo.dnsproxyserver.utils.InetAddresses;
 import dagger.Module;
 import dagger.Provides;
-import org.xbill.DNS.Resolver;
-import org.xbill.DNS.SimpleResolver;
 
 import javax.enterprise.inject.Produces;
 import java.time.Duration;
diff --git a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/RemoteResolvers.java b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/RemoteResolvers.java
index ba28a6b3..e566143d 100644
--- a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/RemoteResolvers.java
+++ b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/RemoteResolvers.java
@@ -1,7 +1,6 @@
 package com.mageddo.dnsproxyserver.server.dns.solver;
 
 import com.mageddo.net.IpAddr;
-import org.xbill.DNS.Resolver;
 
 import java.util.List;
 import java.util.function.Function;
@@ -10,7 +9,7 @@ public class RemoteResolvers {
 
   private final List<Resolver> resolvers;
 
-  public RemoteResolvers(List<Resolver> resolvers) {
+  private RemoteResolvers(List<Resolver> resolvers) {
     this.resolvers = resolvers;
   }
 
diff --git a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/Resolver.java b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/Resolver.java
new file mode 100644
index 00000000..d5012c1f
--- /dev/null
+++ b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/Resolver.java
@@ -0,0 +1,8 @@
+package com.mageddo.dnsproxyserver.server.dns.solver;
+
+
+import java.net.InetSocketAddress;
+
+public interface Resolver extends org.xbill.DNS.Resolver {
+  InetSocketAddress getAddress();
+}
diff --git a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SimpleResolver.java b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SimpleResolver.java
new file mode 100644
index 00000000..87f3e9ba
--- /dev/null
+++ b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SimpleResolver.java
@@ -0,0 +1,23 @@
+package com.mageddo.dnsproxyserver.server.dns.solver;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+
+public class SimpleResolver extends org.xbill.DNS.SimpleResolver implements Resolver {
+
+  public SimpleResolver() throws UnknownHostException {
+  }
+
+  public SimpleResolver(String hostname) throws UnknownHostException {
+    super(hostname);
+  }
+
+  public SimpleResolver(InetSocketAddress host) {
+    super(host);
+  }
+
+  public SimpleResolver(InetAddress host) {
+    super(host);
+  }
+}
diff --git a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemote.java b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemote.java
index 2c2758f6..fa2bbeb1 100644
--- a/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemote.java
+++ b/src/main/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemote.java
@@ -1,6 +1,13 @@
 package com.mageddo.dnsproxyserver.server.dns.solver;
 
+import com.mageddo.commons.circuitbreaker.CircuitCheckException;
+import com.mageddo.commons.concurrent.ThreadPool;
+import com.mageddo.commons.concurrent.Threads;
 import com.mageddo.dnsproxyserver.server.dns.Messages;
+import com.mageddo.net.Networks;
+import dev.failsafe.CircuitBreaker;
+import dev.failsafe.CircuitBreakerOpenException;
+import dev.failsafe.Failsafe;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ClassUtils;
@@ -8,12 +15,19 @@ import org.apache.commons.lang3.time.StopWatch;
 import org.xbill.DNS.Flags;
 import org.xbill.DNS.Message;
 import org.xbill.DNS.Rcode;
-import org.xbill.DNS.Resolver;
 
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static com.mageddo.dnsproxyserver.server.dns.Messages.simplePrint;
 
@@ -25,52 +39,145 @@ public class SolverRemote implements Solver {
   public static final Duration DEFAULT_SUCCESS_TTL = Duration.ofMinutes(5);
   public static final Duration DEFAULT_NXDOMAIN_TTL = Duration.ofMinutes(60);
 
+  static final String QUERY_TIMED_OUT_MSG = "Query timed out";
+  static final long FPS_120 = 1000 / 120;
+
   private final RemoteResolvers delegate;
+  private final Map<InetSocketAddress, CircuitBreaker<Response>> circuitBreakerMap = new ConcurrentHashMap<>();
+  private final ExecutorService threadPool = ThreadPool.newFixed(50);
+  private String status;
 
   @Override
   public Response handle(Message query) {
     final var stopWatch = StopWatch.createStarted();
-    Message lastErrorMsg = null;
+    final var lastErrorMsg = new AtomicReference<Message>();
     for (int i = 0; i < this.delegate.resolvers().size(); i++) {
+
       stopWatch.split();
-      final Resolver resolver = this.delegate.resolvers().get(i);
+      final var resolver = this.delegate.resolvers().get(i);
+      final var circuitBreaker = this.circuitBreakerFor(resolver.getAddress());
+      final var idx = i;
+
       try {
+        final var response = Failsafe
+          .with(circuitBreaker)
+          .get((ctx) -> this.handle0(idx, stopWatch, resolver, query, lastErrorMsg));
+        if (response != null) {
+          return response;
+        }
+      } catch (CircuitCheckException | CircuitBreakerOpenException e) {
+        final var clazz = ClassUtils.getSimpleName(e);
+        log.debug("status=circuitEvent, server={}, type={}", resolver.getAddress(), clazz);
+        this.status = String.format("%s for %s", clazz, resolver.getAddress());
+        continue;
+      }
+    }
+    final var hasErrorResult = lastErrorMsg.get() == null;
+    log.debug("status=finally, time={}, hasErrorResult={}", stopWatch.getTime(), hasErrorResult);
+    if (hasErrorResult) {
+      return null;
+    }
+    return Response.of(lastErrorMsg.get(), DEFAULT_NXDOMAIN_TTL);
+  }
 
-        final var res = Messages.setFlag(resolver.send(query), Flags.RA);
+  private Response handle0(
+    int i,
+    StopWatch stopWatch,
+    Resolver resolver,
+    Message query,
+    AtomicReference<Message> lastErrorMsg
+  ) {
 
-        if (res.getRcode() == Rcode.NOERROR) {
-          log.trace(
-            "status=found, i={}, time={}, req={}, res={}, server={}",
-            i, stopWatch.getTime(), simplePrint(query), simplePrint(res), resolver
-          );
-          return Response.of(res, DEFAULT_SUCCESS_TTL);
-        } else {
-          lastErrorMsg = res;
-          log.trace(
-            "status=notFound, i={}, time={}, req={}, res={}, server={}",
-            i, stopWatch.getTime(), simplePrint(query), simplePrint(res), resolver
-          );
-        }
-      } catch (IOException e) {
+    final var resFuture = resolver.sendAsync(query).toCompletableFuture();
+    final var address = resolver.getAddress();
+    final var pingFuture = this.threadPool.submit(() -> Networks.ping(address.getAddress(), address.getPort(), 1_500));
+
+    boolean mustCheckPing = true;
+    while (true) {
+      if (mustCheckPing && pingFuture.isDone()) {
+        testPing(address, pingFuture);
+        mustCheckPing = false;
+      }
+      if (resFuture.isDone()) {
+        return this.treatResponse(i, resFuture, stopWatch, lastErrorMsg, query, resolver);
+      }
+      Threads.sleep(FPS_120);
+    }
+
+  }
+
+  private static void testPing(InetSocketAddress address, Future<Boolean> pingFuture) {
+    try {
+      final var pingSuccess = pingFuture.get();
+      log.info(
+        "stats=pingTested, success={}, address={}:{}", pingSuccess, address.getAddress(), address.getPort()
+      );
+      if (!pingSuccess) {
+        throw new CircuitCheckException(String.format(
+          "Failed to ping: %s:%s", address.getAddress(), address.getPort()
+        ));
+      }
+    } catch (InterruptedException | ExecutionException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private Response treatResponse(
+    int i,
+    CompletableFuture<Message> resFuture,
+    StopWatch stopWatch,
+    AtomicReference<Message> lastErrorMsg,
+    Message query,
+    Resolver resolver) {
+    try {
+      final var res = Messages.setFlag(resFuture.get(), Flags.RA);
+      if (res.getRcode() == Rcode.NOERROR) {
+        log.trace(
+          "status=found, i={}, time={}, req={}, res={}, server={}",
+          i, stopWatch.getTime(), simplePrint(query), simplePrint(res), resolver
+        );
+        return Response.of(res, DEFAULT_SUCCESS_TTL);
+      } else {
+        lastErrorMsg.set(res);
+        log.trace(
+          "status=notFound, i={}, time={}, req={}, res={}, server={}",
+          i, stopWatch.getTime(), simplePrint(query), simplePrint(res), resolver
+        );
+        return null;
+      }
+    } catch (InterruptedException | ExecutionException e) {
+      if (e.getCause() instanceof IOException) {
         final var time = stopWatch.getTime() - stopWatch.getSplitTime();
-        if (e.getMessage().contains("Timed out while trying")) {
+        if (e.getMessage().contains(QUERY_TIMED_OUT_MSG)) {
           log.info(
             "status=timedOut, i={}, time={}, req={}, msg={} class={}",
             i, time, simplePrint(query), e.getMessage(), ClassUtils.getSimpleName(e)
           );
-          continue;
+          throw new CircuitCheckException(e);
         }
         log.warn(
           "status=failed, i={}, time={}, req={}, server={}, errClass={}, msg={}",
           i, time, simplePrint(query), resolver, ClassUtils.getSimpleName(e), e.getMessage(), e
         );
+        return null;
       }
+      throw new RuntimeException(e.getMessage(), e);
     }
-    final var hasErrorResult = lastErrorMsg == null;
-    log.debug("status=finally, time={}, hasErrorResult={}", stopWatch.getTime(), hasErrorResult);
-    if (hasErrorResult) {
-      return null;
-    }
-    return Response.of(lastErrorMsg, DEFAULT_NXDOMAIN_TTL);
+  }
+
+  private CircuitBreaker<Response> circuitBreakerFor(InetSocketAddress address) {
+    return this.circuitBreakerMap.computeIfAbsent(address, inetSocketAddress -> {
+      final var breaker = CircuitBreaker.<Response>builder()
+        .handle(CircuitCheckException.class)
+        .withFailureThreshold(3, 10)
+        .withSuccessThreshold(5)
+        .withDelay(Duration.ofSeconds(20))
+        .build();
+      return breaker;
+    });
+  }
+
+  String getStatus() {
+    return this.status;
   }
 }
diff --git a/src/main/java/com/mageddo/net/Networks.java b/src/main/java/com/mageddo/net/Networks.java
index 97a28c83..b1c9f5c7 100644
--- a/src/main/java/com/mageddo/net/Networks.java
+++ b/src/main/java/com/mageddo/net/Networks.java
@@ -2,11 +2,16 @@ package com.mageddo.net;
 
 import com.github.dockerjava.api.model.Container;
 import com.github.dockerjava.api.model.ContainerNetwork;
+import com.mageddo.dnsproxyserver.utils.Ips;
 import lombok.SneakyThrows;
 import org.apache.commons.lang3.StringUtils;
 
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
 import java.net.NetworkInterface;
+import java.net.Socket;
 import java.net.SocketException;
+import java.net.UnknownHostException;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Optional;
@@ -99,4 +104,23 @@ public class Networks {
       case IPV6 -> findIpv6Address(network);
     };
   }
+
+  public static boolean ping(String ip, int port, int timeout){
+    try {
+      final InetAddress addr = InetAddress.getByAddress(Ips.toBytes(ip));
+      return ping(addr, port, timeout);
+    } catch (UnknownHostException e) {
+      return false;
+    }
+  }
+
+  public static boolean ping(InetAddress address, int port, int timeout) {
+    try (var socket = new Socket()) {
+      socket.connect(new InetSocketAddress(address, port), timeout);
+      return true;
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
 }


Test patch:
diff --git a/src/test/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemoteTest.java b/src/test/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemoteTest.java
index 25c7ec8a..4365cba7 100644
--- a/src/test/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemoteTest.java
+++ b/src/test/java/com/mageddo/dnsproxyserver/server/dns/solver/SolverRemoteTest.java
@@ -1,6 +1,5 @@
 package com.mageddo.dnsproxyserver.server.dns.solver;
 
-import testing.templates.MessageTemplates;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
@@ -8,17 +7,20 @@ import org.mockito.Mock;
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.xbill.DNS.Flags;
-import org.xbill.DNS.Resolver;
+import org.xbill.DNS.Rcode;
+import testing.templates.InetSocketAddressTemplates;
+import testing.templates.MessageTemplates;
 
-import java.io.IOException;
+import java.net.SocketTimeoutException;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.doThrow;
 
 @ExtendWith(MockitoExtension.class)
 class SolverRemoteTest {
@@ -26,6 +28,9 @@ class SolverRemoteTest {
   @Mock
   Resolver resolver;
 
+  @Mock
+  Resolver resolver2;
+
   @Mock
   RemoteResolvers resolvers;
 
@@ -39,11 +44,16 @@ class SolverRemoteTest {
     final var query = MessageTemplates.acmeAQuery();
     final var answer = MessageTemplates.buildAAnswer(query);
 
-    doReturn(answer)
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
       .when(this.resolver)
-      .send(any())
+      .getAddress()
     ;
 
+    doReturn(CompletableFuture.completedFuture(answer))
+      .when(this.resolver)
+      .sendAsync(any());
+
+
     doReturn(List.of(this.resolver))
       .when(this.resolvers)
       .resolvers()
@@ -62,11 +72,16 @@ class SolverRemoteTest {
     final var query = MessageTemplates.acmeAQuery();
     final var answer = MessageTemplates.buildNXAnswer(query);
 
-    doReturn(answer)
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
       .when(this.resolver)
-      .send(any())
+      .getAddress()
     ;
 
+    doReturn(CompletableFuture.completedFuture(answer))
+      .when(this.resolver)
+      .sendAsync(any());
+
+
     doReturn(List.of(this.resolver))
       .when(this.resolvers)
       .resolvers()
@@ -84,11 +99,15 @@ class SolverRemoteTest {
     // arrange
     final var query = MessageTemplates.acmeAQuery();
 
-    doThrow(new IOException("Deu ruim"))
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
       .when(this.resolver)
-      .send(any())
+      .getAddress()
     ;
 
+    doReturn(CompletableFuture.failedFuture(new SocketTimeoutException("Deu ruim")))
+      .when(this.resolver)
+        .sendAsync(any());
+
     doReturn(List.of(this.resolver))
       .when(this.resolvers)
       .resolvers()
@@ -101,7 +120,6 @@ class SolverRemoteTest {
     assertNull(res);
   }
 
-
   @Test
   void mustReturnRaEvenWhenRemoteServerDoesntReturns() throws Exception {
     // arrange
@@ -109,11 +127,15 @@ class SolverRemoteTest {
     final var res = MessageTemplates.buildAAnswer(query);
     res.getHeader().unsetFlag(Flags.RA);
 
-    doReturn(res)
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
       .when(this.resolver)
-      .send(any())
+      .getAddress()
     ;
 
+    doReturn(CompletableFuture.completedFuture(res))
+      .when(this.resolver)
+      .sendAsync(any());
+
     doReturn(List.of(this.resolver))
       .when(this.resolvers)
       .resolvers()
@@ -127,4 +149,78 @@ class SolverRemoteTest {
     assertEquals(SolverRemote.DEFAULT_SUCCESS_TTL, result.getTtl());
   }
 
+
+  @Test
+  void mustOpenCircuitAfterThresholdFailures() throws Exception {
+    // arrange
+    final var query = MessageTemplates.acmeAQuery();
+
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
+      .when(this.resolver)
+      .getAddress()
+    ;
+
+    doReturn(CompletableFuture.failedFuture(new SocketTimeoutException(SolverRemote.QUERY_TIMED_OUT_MSG)))
+      .when(this.resolver)
+      .sendAsync(any());
+
+    doReturn(List.of(this.resolver))
+      .when(this.resolvers)
+      .resolvers()
+    ;
+
+    // act
+    // assert
+    assertNull(this.solverRemote.handle(query));
+    assertEquals("CircuitCheckException for /8.8.8.8:53", this.solverRemote.getStatus());
+
+    assertNull(this.solverRemote.handle(query));
+    assertEquals("CircuitCheckException for /8.8.8.8:53", this.solverRemote.getStatus());
+
+    assertNull(this.solverRemote.handle(query));
+    assertEquals("CircuitCheckException for /8.8.8.8:53", this.solverRemote.getStatus());
+
+    assertNull(this.solverRemote.handle(query));
+    assertEquals("CircuitBreakerOpenException for /8.8.8.8:53", this.solverRemote.getStatus());
+
+  }
+
+
+  @Test
+  void mustCheckNextServerAfterCircuitFailure() throws Exception {
+    // arrange
+    final var query = MessageTemplates.acmeAQuery();
+    final var res = MessageTemplates.acmeAResponse();
+
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
+      .when(this.resolver)
+      .getAddress()
+    ;
+
+    doReturn(InetSocketAddressTemplates._8_8_8_8())
+      .when(this.resolver2)
+      .getAddress()
+    ;
+
+    doReturn(CompletableFuture.failedFuture(new SocketTimeoutException(SolverRemote.QUERY_TIMED_OUT_MSG)))
+      .when(this.resolver)
+      .sendAsync(any());
+
+    doReturn(CompletableFuture.completedFuture(res))
+      .when(this.resolver2)
+      .sendAsync(any());
+
+    doReturn(List.of(this.resolver, this.resolver2))
+      .when(this.resolvers)
+      .resolvers()
+    ;
+
+    // act
+    // assert
+    final var msg = this.solverRemote.handle(query);
+    assertNotNull(msg);
+    assertEquals("CircuitCheckException for /8.8.8.8:53", this.solverRemote.getStatus());
+    assertEquals(Rcode.NOERROR, msg.getMessage().getRcode());
+
+  }
 }
diff --git a/src/test/java/testing/templates/InetSocketAddressTemplates.java b/src/test/java/testing/templates/InetSocketAddressTemplates.java
new file mode 100644
index 00000000..8db4bd92
--- /dev/null
+++ b/src/test/java/testing/templates/InetSocketAddressTemplates.java
@@ -0,0 +1,11 @@
+package testing.templates;
+
+import com.mageddo.dnsproxyserver.utils.Ips;
+
+import java.net.InetSocketAddress;
+
+public class InetSocketAddressTemplates {
+  public static InetSocketAddress _8_8_8_8(){
+    return new InetSocketAddress(Ips.toAddress("8.8.8.8"), 53);
+  }
+}