DVLP-012: Add health monitor for nodes

src/main/java/org/cache/Main.java

@@ -1,6 +1,7 @@
 package org.cache;
 
 import org.cache.cluster.CacheNode;
+import org.cache.cluster.ClusterHealthMonitor;
 import org.cache.cluster.routing.ClusterForwardingClient;
 import org.cache.cluster.routing.RoutedCacheService;
 import org.cache.config.CacheConfig;
@@ -40,7 +41,7 @@ public static void main(String[] args) {
         configuredCacheConfig = loadConfiguration();
         var app = new SpringApplication(Main.class);
         app.setDefaultProperties(Map.of(
-                "server.port", configuredCacheConfig.cacheNode().httpPort()
+                "server.port", configuredCacheConfig.cacheNode().getHttpPort()
         ));
         app.run(args);
     }
@@ -139,6 +140,15 @@ public ClusterForwardingClient clusterForwardingClient() {
         return new ClusterForwardingClient();
     }
 
+    @Bean
+    public ClusterHealthMonitor clusterHealthMonitor(
+            CacheNode cacheNode,
+            CacheConfig cacheConfig,
+            ClusterForwardingClient forwardingClient
+    ) {
+        return new ClusterHealthMonitor(cacheNode, cacheConfig.clusterInfo(), forwardingClient);
+    }
+
     @Bean(destroyMethod = "shutdownNow")
     public ExecutorService tcpClientExecutor() {
         return Executors.newFixedThreadPool(SERVER_THREAD_COUNT);
@@ -156,7 +166,7 @@ public TcpCacheServer clientTcpCacheServer(
             @Qualifier("commandProcessor") CommandProcessor<Object> commandProcessor
     ) throws IOException {
         return new TcpCacheServer(
-                new ServerSocket(cacheNode.tcpPort()),
+                new ServerSocket(cacheNode.getTcpPort()),
                 tcpClientExecutor,
                 socket -> new ClientConnectionHandler(socket, commandProcessor::process)
         );
@@ -169,7 +179,7 @@ public TcpCacheServer clusterTcpCacheServer(
             @Qualifier("clusterCommandProcessor") CommandProcessor<Object> clusterCommandProcessor
     ) throws IOException {
         return new TcpCacheServer(
-                new ServerSocket(cacheNode.clusterPort()),
+                new ServerSocket(cacheNode.getClusterPort()),
                 clusterClientExecutor,
                 socket -> new ClientConnectionHandler(socket, clusterCommandProcessor::process)
         );

src/main/java/org/cache/cluster/CacheNode.java

@@ -1,15 +1,83 @@
 package org.cache.cluster;
 
-public record CacheNode(
-        String id,
-        String host,
-        int httpPort,
-        int tcpPort,
-        int clusterPort,
-        NodeStatus status
-) {
+import java.util.Objects;
+
+public final class CacheNode {
+
+    private final String id;
+
+    private final String host;
+
+    private final int httpPort;
+
+    private final int tcpPort;
+
+    private final int clusterPort;
+
+    private volatile NodeStatus status;
 
     public CacheNode(String id, String host, int httpPort, int tcpPort, int clusterPort) {
         this(id, host, httpPort, tcpPort, clusterPort, NodeStatus.HEALTHY);
     }
+
+    public CacheNode(String id, String host, int httpPort, int tcpPort, int clusterPort, NodeStatus status) {
+        this.id = Objects.requireNonNull(id, "Node id must not be null");
+        this.host = Objects.requireNonNull(host, "Node host must not be null");
+        this.httpPort = httpPort;
+        this.tcpPort = tcpPort;
+        this.clusterPort = clusterPort;
+        this.status = Objects.requireNonNull(status, "Node status must not be null");
+    }
+
+    public String getId() {
+        return id;
+    }
+
+    public String getHost() {
+        return host;
+    }
+
+    public int getHttpPort() {
+        return httpPort;
+    }
+
+    public int getTcpPort() {
+        return tcpPort;
+    }
+
+    public int getClusterPort() {
+        return clusterPort;
+    }
+
+    public NodeStatus getStatus() {
+        return status;
+    }
+
+    public void setStatus(NodeStatus status) {
+        this.status = Objects.requireNonNull(status, "Node status must not be null");
+    }
+
+    @Override
+    public boolean equals(Object object) {
+        if (this == object) {
+            return true;
+        }
+
+        if (!(object instanceof CacheNode cacheNode)) {
+            return false;
+        }
+
+        return httpPort == cacheNode.httpPort
+                && tcpPort == cacheNode.tcpPort
+                && clusterPort == cacheNode.clusterPort
+                && id.equals(cacheNode.id)
+                && host.equals(cacheNode.host)
+                && status == cacheNode.status;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(id, host, httpPort, tcpPort, clusterPort, status);
+    }
+
 }

src/main/java/org/cache/cluster/ClusterHealthMonitor.java

@@ -0,0 +1,115 @@
+package org.cache.cluster;
+
+import org.cache.cluster.routing.ClusterForwardingClient;
+import org.springframework.context.SmartLifecycle;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+public class ClusterHealthMonitor implements SmartLifecycle {
+
+    private static final int SUSPECTED_FAILURE_COUNT = 2;
+    private static final int UNAVAILABLE_FAILURE_COUNT = 3;
+    private static final long INITIAL_DELAY_SECONDS = 5;
+    private static final long CHECK_INTERVAL_SECONDS = 5;
+
+    private final CacheNode currentNode;
+    private final ClusterInfo clusterInfo;
+    private final ClusterForwardingClient forwardingClient;
+    private final Map<String, Integer> failureCounts = new HashMap<>();
+    private final ScheduledExecutorService executor;
+    private volatile boolean running;
+
+    public ClusterHealthMonitor(
+            CacheNode currentNode,
+            ClusterInfo clusterInfo,
+            ClusterForwardingClient forwardingClient
+    ) {
+        this(currentNode, clusterInfo, forwardingClient, Executors.newSingleThreadScheduledExecutor());
+    }
+
+    ClusterHealthMonitor(
+            CacheNode currentNode,
+            ClusterInfo clusterInfo,
+            ClusterForwardingClient forwardingClient,
+            ScheduledExecutorService executor
+    ) {
+        this.currentNode = currentNode;
+        this.clusterInfo = clusterInfo;
+        this.forwardingClient = forwardingClient;
+        this.executor = executor;
+    }
+
+    @Override
+    public void start() {
+        if (running || clusterInfo == null) {
+            return;
+        }
+
+        running = true;
+        executor.scheduleAtFixedRate(
+                this::checkClusterSafely,
+                INITIAL_DELAY_SECONDS,
+                CHECK_INTERVAL_SECONDS,
+                TimeUnit.SECONDS
+        );
+    }
+
+    @Override
+    public void stop() {
+        running = false;
+        executor.shutdownNow();
+    }
+
+    @Override
+    public boolean isRunning() {
+        return running;
+    }
+
+    void checkCluster() {
+        if (clusterInfo == null) {
+            return;
+        }
+
+        for (CacheNode node : clusterInfo.nodes()) {
+            if (!node.getId().equals(currentNode.getId())) {
+                checkNode(node);
+            }
+        }
+    }
+
+    private void checkClusterSafely() {
+        try {
+            checkCluster();
+        } catch (RuntimeException exception) {
+            System.err.println("Cluster health check failed: " + exception.getMessage());
+        }
+    }
+
+    private void checkNode(CacheNode node) {
+        if (forwardingClient.ping(node)) {
+            failureCounts.remove(node.getId());
+            node.setStatus(NodeStatus.HEALTHY);
+            return;
+        }
+
+        int failures = failureCounts.getOrDefault(node.getId(), 0) + 1;
+        failureCounts.put(node.getId(), failures);
+        node.setStatus(statusFor(failures));
+    }
+
+    private NodeStatus statusFor(int failures) {
+        if (failures >= UNAVAILABLE_FAILURE_COUNT) {
+            return NodeStatus.UNAVAILABLE;
+        }
+
+        if (failures == SUSPECTED_FAILURE_COUNT) {
+            return NodeStatus.SUSPECTED;
+        }
+
+        return NodeStatus.HEALTHY;
+    }
+}

src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java

@@ -2,6 +2,7 @@
 
 import org.cache.cluster.CacheNode;
 import org.cache.cluster.ClusterInfo;
+import org.cache.cluster.NodeStatus;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
@@ -40,7 +41,12 @@ public ConsistentHashRing(ClusterInfo clusterInfo, int virtualNodeCount) {
     }
 
     public CacheNode nodeFor(String key) {
-        return nodesFor(key).getFirst();
+        List<CacheNode> nodes = nodesFor(key);
+        if (nodes.isEmpty()) {
+            throw new IllegalStateException("No available nodes in hash ring");
+        }
+
+        return nodes.getFirst();
     }
 
     public List<CacheNode> nodesFor(String key) {
@@ -64,13 +70,13 @@ private void buildRing(List<CacheNode> nodes, int virtualNodeCount) {
 
     private void addNode(CacheNode node, int virtualNodeCount) {
         for (int index = 0; index < virtualNodeCount; index++) {
-            ring.put(hash(node.id() + "#" + index), node);
+            ring.put(hash(node.getId() + "#" + index), node);
         }
     }
 
     private void addNodesFromRing(long startHash, List<CacheNode> nodes, Set<String> selectedNodeIds) {
         for (CacheNode node : ring.tailMap(startHash, true).values()) {
-            if (selectedNodeIds.add(node.id())) {
+            if (node.getStatus() != NodeStatus.UNAVAILABLE && selectedNodeIds.add(node.getId())) {
                 nodes.add(node);
             }
 

src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java

@@ -8,6 +8,9 @@
 
 public class ClusterForwardingClient {
 
+    private static final List<String> PING_COMMAND = List.of("PING");
+    private static final List<String> PONG_RESPONSE = List.of("PONG");
+
     private final RespCommandClient commandClient;
 
     public ClusterForwardingClient() {
@@ -20,9 +23,17 @@ public ClusterForwardingClient() {
 
     public List<String> forward(CacheNode targetNode, List<String> commandParts) {
         try {
-            return commandClient.send(targetNode.host(), targetNode.clusterPort(), commandParts);
+            return commandClient.send(targetNode.getHost(), targetNode.getClusterPort(), commandParts);
+        } catch (IOException exception) {
+            throw new ClusterForwardingException("Failed to forward request to node: " + targetNode.getId(), exception);
+        }
+    }
+
+    public boolean ping(CacheNode targetNode) {
+        try {
+            return PONG_RESPONSE.equals(commandClient.send(targetNode.getHost(), targetNode.getClusterPort(), PING_COMMAND));
         } catch (IOException exception) {
-            throw new ClusterForwardingException("Failed to forward request to node: " + targetNode.id(), exception);
+            return false;
         }
     }
 }

src/main/java/org/cache/cluster/routing/RoutedCacheService.java

@@ -173,7 +173,7 @@ private ReplicationTargets writeTargetsFor(K key) {
 
         if (!includesCurrentNode && !forwardingAllowed) {
             throw new ClusterForwardingException("Request routed to wrong node. Expected one of owners: "
-                    + ownerIds(owners) + ", current node: " + currentNode.id());
+                    + ownerIds(owners) + ", current node: " + currentNode.getId());
         }
 
         if (!forwardingAllowed) {
@@ -199,7 +199,7 @@ private List<CacheNode> readOwnersFor(K key) {
 
         if (!includesCurrentNode && !forwardingAllowed) {
             throw new ClusterForwardingException("Request routed to wrong node. Expected one of owners: "
-                    + ownerIds(owners) + ", current node: " + currentNode.id());
+                    + ownerIds(owners) + ", current node: " + currentNode.getId());
         }
 
         if (!forwardingAllowed) {
@@ -245,12 +245,12 @@ private Optional<List<String>> remoteLrange(CacheNode owner, K key, int from, in
     }
 
     private boolean isCurrentNode(CacheNode owner) {
-        return owner.id().equals(currentNode.id());
+        return owner.getId().equals(currentNode.getId());
     }
 
     private String ownerIds(List<CacheNode> owners) {
         return owners.stream()
-                .map(CacheNode::id)
+                .map(CacheNode::getId)
                 .toList()
                 .toString();
     }

src/main/java/org/cache/config/CacheConfigLoader.java

@@ -140,13 +140,13 @@ private void validateClusterInfo(int replicationFactor, List<CacheNode> nodes) {
         Set<String> hostPorts = new HashSet<>();
 
         for (CacheNode node : nodes) {
-            if (!nodeIds.add(node.id())) {
-                throw new CacheConfigException("Cluster node ids must be unique: " + node.id());
+            if (!nodeIds.add(node.getId())) {
+                throw new CacheConfigException("Cluster node ids must be unique: " + node.getId());
             }
 
-            addHostPort(hostPorts, node.host(), node.httpPort());
-            addHostPort(hostPorts, node.host(), node.tcpPort());
-            addHostPort(hostPorts, node.host(), node.clusterPort());
+            addHostPort(hostPorts, node.getHost(), node.getHttpPort());
+            addHostPort(hostPorts, node.getHost(), node.getTcpPort());
+            addHostPort(hostPorts, node.getHost(), node.getClusterPort());
         }
     }