More Effective RemoteSolver Circuit Check #526 (#528)

* refactoring

* release notes, Log Remote Servers circuit states

* [Gradle Release Plugin] - new version commit:  '3.25.4-snapshot'.

* ignoring to half open and from half open to open transition

* refactoring

* refactoring

* comment

* add logs

* fixme notes

* comments

* adjusting fixme notes

* docs

* clean code

Author: mageddo

.gitignore

@@ -6,7 +6,7 @@
 Thumbs.db
 
 # Build Files
-*.log
+*.log*
 bin
 target
 build/

RELEASE-NOTES.md

@@ -1,3 +1,8 @@
+## 3.25.4
+* Disable CircuitBreaker check by using ping, keep using the queries to check the server status #526
+* Disable Cache Flush when Circuit Breaker change state to Half Open #526
+* Log Remote Servers circuit states #526
+
 ## 3.25.3
 * Resources utilization optimization by disabling connection check at every query #524
 

gradle.properties

@@ -1 +1 @@
-version=3.25.3-snapshot
+version=3.25.4-snapshot

src/main/java/com/mageddo/dnsproxyserver/solver/SolverCacheFactory.java

@@ -72,9 +72,17 @@ public Map<String, Integer> findInstancesSizeMap(Name name) {
       ;
   }
 
+  /**
+   * This method should be called from one single thread, or it will cause deadlock.
+   */
   public void clearCaches() {
+    // fixme #526 possible solutions for the deadlock:
+    //       1 - only one thread can clear the cache at a time
+    //       2 - move the locks to one centralized thread responsible for the cache management
     for (final var cache : this.getCaches()) {
+      log.trace("status=clearing, cache={}", cache.name());
       cache.clear();
+      log.trace("status=cleared, cache={}", cache.name());
     }
   }
 }

src/main/java/com/mageddo/dnsproxyserver/solver/SolverRemote.java

@@ -54,7 +54,7 @@ public Response handle(Message query) {
 
   Result queryResultFromAvailableResolvers(Message query, StopWatch stopWatch) {
     final var lastErrorMsg = new AtomicReference<Message>();
-
+    // fixme #526 better to exclude open circuits.
     for (int i = 0; i < this.delegate.resolvers().size(); i++) {
 
       final var resolver = this.delegate.resolvers().get(i);

src/main/java/com/mageddo/dnsproxyserver/solver/remote/application/CircuitBreakerCheckerService.java

@@ -43,6 +43,10 @@ void check(InetSocketAddress server, CircuitBreaker<Result> circuitBreaker) {
       });
   }
 
+  /**
+   * Note: Ping isn't being effective for DPS circuit breaker usage.
+   * @see https://github.com/mageddo/dns-proxy-server/issues/526#issuecomment-2261421618
+   */
   boolean ping(InetSocketAddress server) {
     return Networks.ping(server, SolverRemote.PING_TIMEOUT_IN_MS);
   }

src/main/java/com/mageddo/dnsproxyserver/solver/remote/application/CircuitBreakerFactory.java

@@ -7,17 +7,21 @@
 import com.mageddo.dnsproxyserver.solver.remote.dataprovider.SolverConsistencyGuaranteeDAO;
 import com.mageddo.dnsproxyserver.solver.remote.mapper.CircuitBreakerStateMapper;
 import dev.failsafe.CircuitBreaker;
+import dev.failsafe.Failsafe;
 import dev.failsafe.event.CircuitBreakerStateChangedEvent;
 import dev.failsafe.event.EventListener;
 import lombok.RequiredArgsConstructor;
+import lombok.Value;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.time.StopWatch;
 
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import java.net.InetSocketAddress;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
 
 @Slf4j
 @Singleton
@@ -29,7 +33,14 @@ public class CircuitBreakerFactory {
   private final CircuitBreakerCheckerService circuitBreakerCheckerService;
   private final SolverConsistencyGuaranteeDAO solverConsistencyGuaranteeDAO;
 
-  public CircuitBreaker<Result> createCircuitBreakerFor(InetSocketAddress address) {
+  public Result check(InetSocketAddress remoteAddress, Supplier<Result> sup) {
+    final var circuitBreaker = this.createOrGetCircuitBreaker(remoteAddress);
+    return Failsafe
+      .with(circuitBreaker)
+      .get((ctx) -> sup.get());
+  }
+
+  CircuitBreaker<Result> createOrGetCircuitBreaker(InetSocketAddress address) {
     final var config = this.findCircuitBreakerConfig();
     return this.circuitBreakerMap.computeIfAbsent(address, addr -> buildCircuitBreaker(addr, config));
   }
@@ -44,21 +55,34 @@ CircuitBreaker<Result> buildCircuitBreaker(
       .withDelay(config.getTestDelay())
       .onClose(build("CLOSED", address))
       .onOpen(build("OPEN", address))
-      .onHalfOpen(build("HALF_OPEN", address))
       .build();
   }
 
   EventListener<CircuitBreakerStateChangedEvent> build(String actualStateName, InetSocketAddress address) {
     return event -> {
       final var previousStateName = CircuitBreakerStateMapper.toStateNameFrom(event);
-      this.solverConsistencyGuaranteeDAO.flushCachesFromCircuitBreakerStateChange();
+      if (isHalfOpenToOpen(previousStateName, actualStateName)) {
+        log.trace("status=ignoredTransition, from={}, to={}", previousStateName, actualStateName);
+        return;
+      }
+      log.trace(
+        "status=beforeFlushCaches, address={}, previous={}, actual={}", address, previousStateName, actualStateName
+      );
+      this.flushCache();
       log.debug(
-        "status=clearedCache, address={}, previousStateName={}, actualStateName={}",
-        address, previousStateName, actualStateName
+        "status=clearedCache, address={}, previous={}, actual={}", address, previousStateName, actualStateName
       );
     };
   }
 
+  private static boolean isHalfOpenToOpen(String previousStateName, String actualStateName) {
+    return "HALF_OPEN".equals(previousStateName) && "OPEN".equals(actualStateName);
+  }
+
+  void flushCache() {
+    this.solverConsistencyGuaranteeDAO.flushCachesFromCircuitBreakerStateChange();
+  }
+
   com.mageddo.dnsproxyserver.config.CircuitBreaker findCircuitBreakerConfig() {
     return this.configService.findCurrentConfig()
       .getSolverRemote()
@@ -87,7 +111,33 @@ boolean circuitBreakerSafeCheck(Map.Entry<InetSocketAddress, CircuitBreaker<Resu
     return this.circuitBreakerCheckerService.safeCheck(entry.getKey(), entry.getValue());
   }
 
-  public void reset(){
+  public void reset() {
     this.circuitBreakerMap.clear();
   }
+
+  public List<Stats> stats() {
+    return this.circuitBreakerMap.keySet()
+      .stream()
+      .map(this::toStats)
+      .toList();
+  }
+
+  private Stats toStats(InetSocketAddress remoteAddr) {
+    final var circuitBreaker = this.circuitBreakerMap.get(remoteAddr);
+    final var state = circuitBreaker.getState().name();
+    return Stats.of(remoteAddr.toString(), state);
+  }
+
+
+  @Value
+  public static class Stats {
+
+    private String remoteServerAddress;
+    private String state;
+
+    public static Stats of(String remoteServerAddress, String state) {
+      return new Stats(remoteServerAddress, state);
+    }
+  }
+
 }

src/main/java/com/mageddo/dnsproxyserver/solver/remote/application/CircuitBreakerFailSafeService.java

@@ -4,7 +4,6 @@
 import com.mageddo.dnsproxyserver.solver.remote.CircuitBreakerService;
 import com.mageddo.dnsproxyserver.solver.remote.Result;
 import dev.failsafe.CircuitBreakerOpenException;
-import dev.failsafe.Failsafe;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.ClassUtils;
@@ -36,10 +35,7 @@ public Result safeHandle(InetSocketAddress resolverAddress, Supplier<Result> sup
   }
 
   private Result handle(InetSocketAddress resolverAddress, Supplier<Result> sup) {
-    final var circuitBreaker = this.circuitBreakerFactory.createCircuitBreakerFor(resolverAddress);
-    return Failsafe
-      .with(circuitBreaker)
-      .get((ctx) -> sup.get());
+    return this.circuitBreakerFactory.check(resolverAddress, sup);
   }
 
   public String getStatus() {

src/main/java/com/mageddo/dnsproxyserver/solver/remote/configurator/CircuitBreakerWatchDogScheduler.java

@@ -4,12 +4,14 @@
 import com.mageddo.dnsproxyserver.di.StartupEvent;
 import com.mageddo.dnsproxyserver.solver.remote.application.CircuitBreakerFactory;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+@Slf4j
 @Singleton
 @RequiredArgsConstructor(onConstructor = @__({@Inject}))
 public class CircuitBreakerWatchDogScheduler implements StartupEvent {
@@ -19,6 +21,14 @@ public class CircuitBreakerWatchDogScheduler implements StartupEvent {
 
   @Override
   public void onStart() {
-    this.executor.scheduleWithFixedDelay(this.circuitBreakerFactory::checkCreatedCircuits, 0, 10, TimeUnit.SECONDS);
+    this.executor.scheduleWithFixedDelay(this::logStats, 0, 10, TimeUnit.SECONDS);
+  }
+
+  void logStats() {
+    this.circuitBreakerFactory
+      .stats()
+      .forEach(stats -> {
+        log.debug("stats={}", stats);
+      });
   }
 }

src/test/java/com/mageddo/dnsproxyserver/solver/remote/application/CircuitBreakerFactoryTest.java

@@ -1,16 +1,26 @@
 package com.mageddo.dnsproxyserver.solver.remote.application;
 
+import com.mageddo.commons.circuitbreaker.CircuitCheckException;
+import com.mageddo.commons.concurrent.Threads;
+import com.mageddo.dnsproxyserver.solver.remote.Result;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.InjectMocks;
 import org.mockito.Spy;
 import org.mockito.junit.jupiter.MockitoExtension;
 import testing.templates.CircuitBreakerConfigTemplates;
 import testing.templates.InetSocketAddressTemplates;
+import testing.templates.solver.remote.ResultSupplierTemplates;
+
+import java.net.InetSocketAddress;
+import java.util.function.Supplier;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 @ExtendWith(MockitoExtension.class)
 class CircuitBreakerFactoryTest {
@@ -29,7 +39,7 @@ void mustCheckAllExistentCircuitsAndCountSuccessWhenSafeCheckReturnsTrue() {
     doReturn(true).when(this.factory).circuitBreakerSafeCheck(any());
 
     final var addr = InetSocketAddressTemplates._8_8_8_8();
-    this.factory.createCircuitBreakerFor(addr);
+    this.factory.createOrGetCircuitBreaker(addr);
 
     // act
     final var result = this.factory.checkCreatedCircuits();
@@ -50,7 +60,7 @@ void mustCheckAndCountErrorWhenSafeCheckReturnsFalse() {
     doReturn(false).when(this.factory).circuitBreakerSafeCheck(any());
 
     final var addr = InetSocketAddressTemplates._8_8_8_8();
-    this.factory.createCircuitBreakerFor(addr);
+    this.factory.createOrGetCircuitBreaker(addr);
 
     // act
     final var result = this.factory.checkCreatedCircuits();
@@ -59,4 +69,42 @@ void mustCheckAndCountErrorWhenSafeCheckReturnsFalse() {
     assertEquals(0, result.getKey());
     assertEquals(1, result.getValue());
   }
+
+  @Test
+  void mustNotFlushCacheWhenChangeStateToHalfOpen(){
+
+    // arrange
+    assertEquals("[]", this.factory.stats().toString());
+
+    final var addr = InetSocketAddressTemplates._8_8_8_8();
+    final var supError = ResultSupplierTemplates.alwaysFail();
+    final var supSuccess = ResultSupplierTemplates.alwaysSuccess();
+
+    doReturn(CircuitBreakerConfigTemplates.oneTryFailSuccess())
+      .when(this.factory)
+      .findCircuitBreakerConfig()
+    ;
+
+    this.checkFailAndSleep(addr, supError);
+    this.checkFailAndSleep(addr, supError);
+
+    this.factory.check(addr, supSuccess);
+    assertEquals(
+      "[CircuitBreakerFactory.Stats(remoteServerAddress=/8.8.8.8:53, state=CLOSED)]",
+      this.factory.stats().toString()
+    );
+    verify(this.factory, times(2)).flushCache();
+
+
+  }
+
+  void checkFailAndSleep(InetSocketAddress addr, Supplier<Result> supError) {
+    assertThrows(CircuitCheckException.class, () -> this.factory.check(addr, supError));
+    assertEquals(
+      "[CircuitBreakerFactory.Stats(remoteServerAddress=/8.8.8.8:53, state=OPEN)]",
+      this.factory.stats().toString()
+    );
+    verify(this.factory).flushCache();
+    Threads.sleep(100);
+  }
 }