diff --git a/src/main/java/org/cache/Main.java b/src/main/java/org/cache/Main.java index 4260ff5..a24ef96 100644 --- a/src/main/java/org/cache/Main.java +++ b/src/main/java/org/cache/Main.java @@ -1,7 +1,11 @@ package org.cache; import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterGossipService; import org.cache.cluster.ClusterHealthMonitor; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterMembershipClient; +import org.cache.cluster.ClusterTopology; import org.cache.cluster.routing.ClusterForwardingClient; import org.cache.cluster.routing.RoutedCacheService; import org.cache.config.CacheConfig; @@ -70,6 +74,20 @@ public CacheNode cacheNode(CacheConfig cacheConfig) { return cacheConfig.cacheNode(); } + @Bean + public ClusterMembership clusterMembership(CacheConfig cacheConfig) { + if (cacheConfig.clusterInfo() == null) { + return new ClusterMembership(null); + } + + return new ClusterMembership(new ClusterTopology( + 0, + cacheConfig.clusterInfo().nodes(), + cacheConfig.clusterInfo().replicationFactor(), + 128 + )); + } + @Bean public KeyCodec keyCodec(CacheConfig cacheConfig) { return cacheConfig.keyCodec(); @@ -97,42 +115,49 @@ public CacheService cacheService(Cache cache, ValueCodecRegistry public CacheOperations routedCacheService( CacheService cacheService, CacheNode cacheNode, - CacheConfig cacheConfig, ClusterForwardingClient forwardingClient, - KeyCodec keyCodec + KeyCodec keyCodec, + ClusterMembership clusterMembership ) { - return new RoutedCacheService<>(cacheService, cacheNode, cacheConfig.clusterInfo(), forwardingClient, keyCodec); + return new RoutedCacheService<>(cacheService, cacheNode, forwardingClient, keyCodec, true, clusterMembership); } @Bean public CacheOperations clusterRoutedCacheService( CacheService cacheService, CacheNode cacheNode, - CacheConfig cacheConfig, ClusterForwardingClient forwardingClient, - KeyCodec keyCodec + KeyCodec keyCodec, + ClusterMembership clusterMembership ) { return new RoutedCacheService<>( cacheService, cacheNode, - cacheConfig.clusterInfo(), forwardingClient, keyCodec, - false + false, + clusterMembership ); } @Bean - public CommandProcessor commandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { - return new CommandProcessor<>(keyCodec, cacheService); + public CommandProcessor commandProcessor( + KeyCodec keyCodec, + CacheOperations cacheService, + ClusterMembership clusterMembership, + ClusterGossipService clusterGossipService + ) { + return new CommandProcessor<>(keyCodec, cacheService, clusterMembership, clusterGossipService, false); } @Bean public CommandProcessor clusterCommandProcessor( KeyCodec keyCodec, - @Qualifier("clusterRoutedCacheService") CacheOperations cacheService + @Qualifier("clusterRoutedCacheService") CacheOperations cacheService, + ClusterMembership clusterMembership, + ClusterGossipService clusterGossipService ) { - return new CommandProcessor<>(keyCodec, cacheService); + return new CommandProcessor<>(keyCodec, cacheService, clusterMembership, clusterGossipService); } @Bean @@ -140,13 +165,27 @@ public ClusterForwardingClient clusterForwardingClient() { return new ClusterForwardingClient(); } + @Bean + public ClusterMembershipClient clusterMembershipClient(ClusterForwardingClient forwardingClient) { + return new ClusterMembershipClient(forwardingClient); + } + @Bean public ClusterHealthMonitor clusterHealthMonitor( CacheNode cacheNode, - CacheConfig cacheConfig, - ClusterForwardingClient forwardingClient + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient + ) { + return new ClusterHealthMonitor(cacheNode, clusterMembership, membershipClient); + } + + @Bean + public ClusterGossipService clusterGossipService( + CacheNode cacheNode, + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient ) { - return new ClusterHealthMonitor(cacheNode, cacheConfig.clusterInfo(), forwardingClient); + return new ClusterGossipService(cacheNode, clusterMembership, membershipClient); } @Bean(destroyMethod = "shutdownNow") diff --git a/src/main/java/org/cache/cluster/CacheInfoException.java b/src/main/java/org/cache/cluster/CacheInfoException.java new file mode 100644 index 0000000..73110eb --- /dev/null +++ b/src/main/java/org/cache/cluster/CacheInfoException.java @@ -0,0 +1,7 @@ +package org.cache.cluster; + +public class CacheInfoException extends IllegalArgumentException { + public CacheInfoException(String message) { + super(message); + } +} diff --git a/src/main/java/org/cache/cluster/ClusterGossipService.java b/src/main/java/org/cache/cluster/ClusterGossipService.java new file mode 100644 index 0000000..5aa8a0b --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterGossipService.java @@ -0,0 +1,132 @@ +package org.cache.cluster; + +import org.springframework.context.SmartLifecycle; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public class ClusterGossipService implements SmartLifecycle { + + private static final long INITIAL_DELAY_SECONDS = 3; + private static final long GOSSIP_INTERVAL_SECONDS = 3; + + private final CacheNode currentNode; + private final ClusterMembership clusterMembership; + private final ClusterMembershipClient membershipClient; + private final ScheduledExecutorService executor; + private volatile boolean running; + + public ClusterGossipService( + CacheNode currentNode, + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient + ) { + this(currentNode, clusterMembership, membershipClient, Executors.newSingleThreadScheduledExecutor()); + } + + ClusterGossipService( + CacheNode currentNode, + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient, + ScheduledExecutorService executor + ) { + this.currentNode = currentNode; + this.clusterMembership = clusterMembership; + this.membershipClient = membershipClient; + this.executor = executor; + } + + @Override + public void start() { + if (running || clusterMembership.currentTopology() == null) { + return; + } + + running = true; + executor.scheduleAtFixedRate( + this::gossipOnce, + INITIAL_DELAY_SECONDS, + GOSSIP_INTERVAL_SECONDS, + TimeUnit.SECONDS + ); + } + + @Override + public void stop() { + running = false; + executor.shutdownNow(); + } + + @Override + public boolean isRunning() { + return running; + } + + public void broadcastTopology() { + ClusterTopology topology = clusterMembership.currentTopology(); + if (topology == null) { + return; + } + + for (CacheNode peer : topology.nodes()) { + if (isNotCurrentNode(peer)) { + membershipClient.applyTopology(peer, topology); + } + } + } + + void gossipOnce() { + try { + ClusterTopology topology = clusterMembership.currentTopology(); + if (topology == null) { + return; + } + + for (CacheNode peer : topology.nodes()) { + if (isNotCurrentNode(peer)) { + gossipWith(peer); + } + } + } catch (RuntimeException exception) { + System.err.println("Cluster gossip failed: " + exception.getMessage()); + } + + } + + private void gossipWith(CacheNode peer) { + ClusterTopology localTopology = clusterMembership.currentTopology(); + membershipClient.topologyDigest(peer).ifPresent(peerDigest -> { + if (peerDigest.version() < localTopology.version()) { + membershipClient.applyTopology(peer, localTopology); + return; + } + + if (peerDigest.version() > localTopology.version()) { + applyPeerTopology(peer); + return; + } + + if (!peerDigest.fingerprint().equals(localTopology.fingerprint())) { + resolveSameVersionConflict(peer, localTopology); + } + }); + } + + private void resolveSameVersionConflict(CacheNode peer, ClusterTopology localTopology) { + boolean appliedPeerTopology = applyPeerTopology(peer); + if (!appliedPeerTopology) { + membershipClient.applyTopology(peer, localTopology); + } + } + + private boolean applyPeerTopology(CacheNode peer) { + return membershipClient.topology(peer) + .map(clusterMembership::applyTopology) + .orElse(false); + } + + private boolean isNotCurrentNode(CacheNode node) { + return !node.getId().equals(currentNode.getId()); + } +} diff --git a/src/main/java/org/cache/cluster/ClusterHealthMonitor.java b/src/main/java/org/cache/cluster/ClusterHealthMonitor.java index a502f62..345c3bd 100644 --- a/src/main/java/org/cache/cluster/ClusterHealthMonitor.java +++ b/src/main/java/org/cache/cluster/ClusterHealthMonitor.java @@ -1,6 +1,5 @@ package org.cache.cluster; -import org.cache.cluster.routing.ClusterForwardingClient; import org.springframework.context.SmartLifecycle; import java.util.HashMap; @@ -17,35 +16,35 @@ public class ClusterHealthMonitor implements SmartLifecycle { private static final long CHECK_INTERVAL_SECONDS = 5; private final CacheNode currentNode; - private final ClusterInfo clusterInfo; - private final ClusterForwardingClient forwardingClient; + private final ClusterMembership clusterMembership; + private final ClusterMembershipClient membershipClient; private final Map failureCounts = new HashMap<>(); private final ScheduledExecutorService executor; private volatile boolean running; public ClusterHealthMonitor( CacheNode currentNode, - ClusterInfo clusterInfo, - ClusterForwardingClient forwardingClient + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient ) { - this(currentNode, clusterInfo, forwardingClient, Executors.newSingleThreadScheduledExecutor()); + this(currentNode, clusterMembership, membershipClient, Executors.newSingleThreadScheduledExecutor()); } ClusterHealthMonitor( CacheNode currentNode, - ClusterInfo clusterInfo, - ClusterForwardingClient forwardingClient, + ClusterMembership clusterMembership, + ClusterMembershipClient membershipClient, ScheduledExecutorService executor ) { this.currentNode = currentNode; - this.clusterInfo = clusterInfo; - this.forwardingClient = forwardingClient; + this.clusterMembership = clusterMembership; + this.membershipClient = membershipClient; this.executor = executor; } @Override public void start() { - if (running || clusterInfo == null) { + if (running || clusterMembership.currentTopology() == null) { return; } @@ -70,11 +69,12 @@ public boolean isRunning() { } void checkCluster() { - if (clusterInfo == null) { + ClusterTopology topology = clusterMembership.currentTopology(); + if (topology == null) { return; } - for (CacheNode node : clusterInfo.nodes()) { + for (CacheNode node : topology.nodes()) { if (!node.getId().equals(currentNode.getId())) { checkNode(node); } @@ -90,15 +90,15 @@ private void checkClusterSafely() { } private void checkNode(CacheNode node) { - if (forwardingClient.ping(node)) { + if (membershipClient.ping(node)) { failureCounts.remove(node.getId()); - node.setStatus(NodeStatus.HEALTHY); + clusterMembership.markStatus(node.getId(), NodeStatus.HEALTHY); return; } int failures = failureCounts.getOrDefault(node.getId(), 0) + 1; failureCounts.put(node.getId(), failures); - node.setStatus(statusFor(failures)); + clusterMembership.markStatus(node.getId(), statusFor(failures)); } private NodeStatus statusFor(int failures) { diff --git a/src/main/java/org/cache/cluster/ClusterMembership.java b/src/main/java/org/cache/cluster/ClusterMembership.java new file mode 100644 index 0000000..739d61c --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterMembership.java @@ -0,0 +1,127 @@ +package org.cache.cluster; + +import org.cache.cluster.hashing.ConsistentHashRing; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Optional; + +public class ClusterMembership { + + private volatile ClusterTopology clusterTopology; + + public ClusterMembership(ClusterTopology clusterTopology) { + this.clusterTopology = clusterTopology; + } + + public ClusterTopology currentTopology() { + return clusterTopology; + } + + public ConsistentHashRing getHashRing() { + if (clusterTopology == null) { + return null; + } + + return clusterTopology.hashRing(); + } + + public synchronized void addNode(CacheNode node) { + if (clusterTopology == null) { + throw new CacheInfoException("Cluster is not enabled"); + } + + List nodes = new ArrayList<>(clusterTopology.nodes()); + nodes.add(node); + ClusterValidator.validateClusterInfo(clusterTopology.replicationFactor(), nodes); + + clusterTopology = nextTopology(nodes); + } + + public synchronized void removeNode(String nodeId) { + if (clusterTopology == null) { + throw new CacheInfoException("Cluster is not enabled"); + } + + if (findNode(nodeId).isEmpty()) { + throw new CacheInfoException("Cluster node does not exist: " + nodeId); + } + + List nodes = clusterTopology.nodes() + .stream() + .filter(node -> !node.getId().equals(nodeId)) + .toList(); + ClusterValidator.validateClusterInfo(clusterTopology.replicationFactor(), nodes); + + clusterTopology = nextTopology(nodes); + } + + public synchronized boolean applyTopology(ClusterTopology incoming) { + if (clusterTopology != null) { + if (incoming.version() < clusterTopology.version()) { + return false; + } + + if (incoming.version() == clusterTopology.version() + && incoming.fingerprint().equals(clusterTopology.fingerprint())) { + return false; + } + + if (incoming.version() == clusterTopology.version() && !hasSmallerTcpAddresses(incoming)) { + return false; + } + } + + ClusterValidator.validateClusterInfo( + incoming.replicationFactor(), + incoming.nodes() + ); + + clusterTopology = new ClusterTopology( + incoming.version(), + incoming.nodes(), + incoming.replicationFactor(), + incoming.virtualNodeCount() + ); + + return true; + } + + public void markStatus(String nodeId, NodeStatus status) { + findNode(nodeId).ifPresent(node -> node.setStatus(status)); + } + + public Optional findNode(String nodeId) { + if (clusterTopology == null) { + return Optional.empty(); + } + + return clusterTopology.nodes() + .stream() + .filter(node -> node.getId().equals(nodeId)) + .findFirst(); + } + + private ClusterTopology nextTopology(List nodes) { + return new ClusterTopology( + clusterTopology.version() + 1, + nodes, + clusterTopology.replicationFactor(), + clusterTopology.virtualNodeCount() + ); + } + + private boolean hasSmallerTcpAddresses(ClusterTopology incoming) { + return tcpAddresses(incoming).compareTo(tcpAddresses(clusterTopology)) < 0; + } + + private String tcpAddresses(ClusterTopology topology) { + return topology.nodes() + .stream() + .map(node -> node.getHost() + ":" + node.getTcpPort()) + .sorted(Comparator.naturalOrder()) + .toList() + .toString(); + } +} diff --git a/src/main/java/org/cache/cluster/ClusterMembershipClient.java b/src/main/java/org/cache/cluster/ClusterMembershipClient.java new file mode 100644 index 0000000..aaf854e --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterMembershipClient.java @@ -0,0 +1,83 @@ +package org.cache.cluster; + +import org.cache.cluster.routing.ClusterForwardingClient; +import org.cache.cluster.routing.ClusterForwardingException; +import org.cache.protocol.handlers.CommandType; +import org.cache.protocol.handlers.ResponseConstants; + +import java.util.List; +import java.util.Optional; + +public class ClusterMembershipClient { + + private static final List PING_COMMAND = List.of("PING"); + private static final List PONG_RESPONSE = List.of("PONG"); + private static final int TOPOLOGY_DIGEST_RESPONSE_SIZE = 3; + private static final int TOPOLOGY_RESPONSE_SIZE = 2; + private static final int TOPOLOGY_VERSION_INDEX = 1; + private static final int TOPOLOGY_FINGERPRINT_INDEX = 2; + private static final int TOPOLOGY_VALUE_INDEX = 1; + + private final ClusterForwardingClient forwardingClient; + private final ClusterTopologyCodec topologyCodec; + + public ClusterMembershipClient(ClusterForwardingClient forwardingClient) { + this(forwardingClient, new ClusterTopologyCodec()); + } + + ClusterMembershipClient(ClusterForwardingClient forwardingClient, ClusterTopologyCodec topologyCodec) { + this.forwardingClient = forwardingClient; + this.topologyCodec = topologyCodec; + } + + public boolean ping(CacheNode targetNode) { + try { + return PONG_RESPONSE.equals(forwardingClient.forward(targetNode, PING_COMMAND)); + } catch (ClusterForwardingException exception) { + return false; + } + } + + public Optional topologyDigest(CacheNode targetNode) { + try { + List response = forwardingClient.forward(targetNode, List.of(CommandType.TOPOLOGY_DIGEST.name())); + if (response.size() != TOPOLOGY_DIGEST_RESPONSE_SIZE + || !CommandType.TOPOLOGY_DIGEST.name().equals(response.getFirst())) { + return Optional.empty(); + } + + return Optional.of(new TopologyDigest( + ClusterNumberParser.parseLong(response.get(TOPOLOGY_VERSION_INDEX), TopologyField.VERSION.getValue()), + response.get(TOPOLOGY_FINGERPRINT_INDEX) + )); + } catch (RuntimeException exception) { + return Optional.empty(); + } + } + + public Optional topology(CacheNode targetNode) { + try { + List response = forwardingClient.forward(targetNode, List.of(CommandType.TOPOLOGY_GET.name())); + if (response.size() != TOPOLOGY_RESPONSE_SIZE + || !ResponseConstants.TOPOLOGY.name().equals(response.getFirst())) { + return Optional.empty(); + } + + return Optional.of(topologyCodec.decode(response.get(TOPOLOGY_VALUE_INDEX))); + } catch (RuntimeException exception) { + return Optional.empty(); + } + } + + public boolean applyTopology(CacheNode targetNode, ClusterTopology topology) { + try { + List response = forwardingClient.forward( + targetNode, + List.of(CommandType.TOPOLOGY_APPLY.name(), topologyCodec.encode(topology)) + ); + return List.of(ResponseConstants.OK.name()).equals(response); + } catch (ClusterForwardingException exception) { + return false; + } + } +} diff --git a/src/main/java/org/cache/cluster/ClusterNumberParser.java b/src/main/java/org/cache/cluster/ClusterNumberParser.java new file mode 100644 index 0000000..c1fe896 --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterNumberParser.java @@ -0,0 +1,23 @@ +package org.cache.cluster; + +public final class ClusterNumberParser { + + private ClusterNumberParser() { + } + + public static int parseInt(String value, String name) { + try { + return Integer.parseInt(value); + } catch (NumberFormatException exception) { + throw new CacheInfoException("Invalid cluster topology " + name + ": " + value); + } + } + + public static long parseLong(String value, String name) { + try { + return Long.parseLong(value); + } catch (NumberFormatException exception) { + throw new CacheInfoException("Invalid cluster topology " + name + ": " + value); + } + } +} diff --git a/src/main/java/org/cache/cluster/ClusterTopology.java b/src/main/java/org/cache/cluster/ClusterTopology.java new file mode 100644 index 0000000..39b071c --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterTopology.java @@ -0,0 +1,51 @@ +package org.cache.cluster; + +import org.cache.cluster.hashing.ConsistentHashRing; +import org.cache.cluster.hashing.ClusterHash; + +import java.util.Comparator; +import java.util.List; + +public record ClusterTopology( + long version, + List nodes, + int replicationFactor, + int virtualNodeCount, + ConsistentHashRing hashRing +) { + + public ClusterTopology(long version, List nodes, int replicationFactor, int virtualNodeCount) { + this( + version, + List.copyOf(nodes), + replicationFactor, + virtualNodeCount, + new ConsistentHashRing(new ClusterInfo(replicationFactor, List.copyOf(nodes)), virtualNodeCount) + ); + } + + public String fingerprint() { + String canonicalTopology = replicationFactor + + "|" + + virtualNodeCount + + "|" + + nodes.stream() + .sorted(Comparator.comparing(CacheNode::getId)) + .map(this::fingerprintNode) + .toList(); + + return ClusterHash.hashHex(canonicalTopology); + } + + private String fingerprintNode(CacheNode node) { + return node.getId() + + "|" + + node.getHost() + + "|" + + node.getHttpPort() + + "|" + + node.getTcpPort() + + "|" + + node.getClusterPort(); + } +} diff --git a/src/main/java/org/cache/cluster/ClusterTopologyCodec.java b/src/main/java/org/cache/cluster/ClusterTopologyCodec.java new file mode 100644 index 0000000..f690f06 --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterTopologyCodec.java @@ -0,0 +1,106 @@ +package org.cache.cluster; + +import java.util.ArrayList; +import java.util.List; + +public class ClusterTopologyCodec { + + private static final String SECTION_SEPARATOR = "|"; + private static final String SECTION_SEPARATOR_REGEX = "\\|"; + private static final String NODE_SEPARATOR = ";"; + private static final String NODE_FIELD_SEPARATOR = ","; + private static final String EMPTY_TOPOLOGY_NODES = ""; + private static final String INVALID_TOPOLOGY_MESSAGE = "Invalid cluster topology"; + private static final String INVALID_NODE_MESSAGE = "Invalid cluster node"; + private static final int HEADER_PARTS = 4; + private static final int VERSION_INDEX = 0; + private static final int REPLICATION_FACTOR_INDEX = 1; + private static final int VIRTUAL_NODE_COUNT_INDEX = 2; + private static final int NODES_INDEX = 3; + private static final int NODE_FIELD_COUNT = 6; + private static final int NODE_ID_INDEX = 0; + private static final int NODE_HOST_INDEX = 1; + private static final int NODE_HTTP_PORT_INDEX = 2; + private static final int NODE_TCP_PORT_INDEX = 3; + private static final int NODE_CLUSTER_PORT_INDEX = 4; + private static final int NODE_STATUS_INDEX = 5; + + public String encode(ClusterTopology topology) { + String nodes = topology.nodes() + .stream() + .map(this::encodeNode) + .reduce((first, second) -> first + NODE_SEPARATOR + second) + .orElse(EMPTY_TOPOLOGY_NODES); + + return topology.version() + + SECTION_SEPARATOR + + topology.replicationFactor() + + SECTION_SEPARATOR + + topology.virtualNodeCount() + + SECTION_SEPARATOR + + nodes; + } + + public ClusterTopology decode(String value) { + String[] parts = value.split(SECTION_SEPARATOR_REGEX, HEADER_PARTS); + if (parts.length != HEADER_PARTS) { + throw new CacheInfoException(INVALID_TOPOLOGY_MESSAGE); + } + + long version = ClusterNumberParser.parseLong(parts[VERSION_INDEX], TopologyField.VERSION.getValue()); + int replicationFactor = ClusterNumberParser.parseInt( + parts[REPLICATION_FACTOR_INDEX], + TopologyField.REPLICATION_FACTOR.getValue() + ); + int virtualNodeCount = ClusterNumberParser.parseInt( + parts[VIRTUAL_NODE_COUNT_INDEX], + TopologyField.VIRTUAL_NODE_COUNT.getValue() + ); + List nodes = decodeNodes(parts[NODES_INDEX]); + + return new ClusterTopology(version, nodes, replicationFactor, virtualNodeCount); + } + + private String encodeNode(CacheNode node) { + return node.getId() + + NODE_FIELD_SEPARATOR + + node.getHost() + + NODE_FIELD_SEPARATOR + + node.getHttpPort() + + NODE_FIELD_SEPARATOR + + node.getTcpPort() + + NODE_FIELD_SEPARATOR + + node.getClusterPort() + + NODE_FIELD_SEPARATOR + + node.getStatus().name(); + } + + private List decodeNodes(String value) { + if (value.isBlank()) { + return List.of(); + } + + List nodes = new ArrayList<>(); + for (String nodeValue : value.split(NODE_SEPARATOR)) { + nodes.add(decodeNode(nodeValue)); + } + + return List.copyOf(nodes); + } + + private CacheNode decodeNode(String value) { + String[] fields = value.split(NODE_FIELD_SEPARATOR, NODE_FIELD_COUNT); + if (fields.length != NODE_FIELD_COUNT) { + throw new CacheInfoException(INVALID_NODE_MESSAGE); + } + + return new CacheNode( + fields[NODE_ID_INDEX], + fields[NODE_HOST_INDEX], + ClusterNumberParser.parseInt(fields[NODE_HTTP_PORT_INDEX], TopologyField.HTTP_PORT.getValue()), + ClusterNumberParser.parseInt(fields[NODE_TCP_PORT_INDEX], TopologyField.TCP_PORT.getValue()), + ClusterNumberParser.parseInt(fields[NODE_CLUSTER_PORT_INDEX], TopologyField.CLUSTER_PORT.getValue()), + NodeStatus.valueOf(fields[NODE_STATUS_INDEX]) + ); + } +} diff --git a/src/main/java/org/cache/cluster/ClusterValidator.java b/src/main/java/org/cache/cluster/ClusterValidator.java new file mode 100644 index 0000000..500a593 --- /dev/null +++ b/src/main/java/org/cache/cluster/ClusterValidator.java @@ -0,0 +1,41 @@ +package org.cache.cluster; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +public final class ClusterValidator { + + private ClusterValidator() { + } + + public static void validateClusterInfo(int replicationFactor, List nodes) { + if (replicationFactor < 1) { + throw new CacheInfoException("Cluster replication factor must be at least 1"); + } + + if (replicationFactor > nodes.size()) { + throw new CacheInfoException("Cluster replication factor must not exceed number of active nodes"); + } + + Set nodeIds = new HashSet<>(); + Set hostPorts = new HashSet<>(); + + for (CacheNode node : nodes) { + if (!nodeIds.add(node.getId())) { + throw new CacheInfoException("Cluster node ids must be unique: " + node.getId()); + } + + addHostPort(hostPorts, node.getHost(), node.getHttpPort()); + addHostPort(hostPorts, node.getHost(), node.getTcpPort()); + addHostPort(hostPorts, node.getHost(), node.getClusterPort()); + } + } + + private static void addHostPort(Set hostPorts, String host, int port) { + String hostPort = host + ":" + port; + if (!hostPorts.add(hostPort)) { + throw new CacheInfoException("Cluster node host-port combinations must be unique: " + hostPort); + } + } +} diff --git a/src/main/java/org/cache/cluster/TopologyDigest.java b/src/main/java/org/cache/cluster/TopologyDigest.java new file mode 100644 index 0000000..4a3294f --- /dev/null +++ b/src/main/java/org/cache/cluster/TopologyDigest.java @@ -0,0 +1,4 @@ +package org.cache.cluster; + +public record TopologyDigest(long version, String fingerprint) { +} diff --git a/src/main/java/org/cache/cluster/TopologyField.java b/src/main/java/org/cache/cluster/TopologyField.java new file mode 100644 index 0000000..49a706f --- /dev/null +++ b/src/main/java/org/cache/cluster/TopologyField.java @@ -0,0 +1,20 @@ +package org.cache.cluster; + +public enum TopologyField { + VERSION("version"), + REPLICATION_FACTOR("replicationFactor"), + VIRTUAL_NODE_COUNT("virtualNodeCount"), + HTTP_PORT("httpPort"), + TCP_PORT("tcpPort"), + CLUSTER_PORT("clusterPort"); + + private final String value; + + TopologyField(String value) { + this.value = value; + } + + String getValue() { + return value; + } +} diff --git a/src/main/java/org/cache/cluster/hashing/ClusterHash.java b/src/main/java/org/cache/cluster/hashing/ClusterHash.java new file mode 100644 index 0000000..9924b68 --- /dev/null +++ b/src/main/java/org/cache/cluster/hashing/ClusterHash.java @@ -0,0 +1,32 @@ +package org.cache.cluster.hashing; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; + +public final class ClusterHash { + + private static final String HASH_ALGORITHM = "SHA-256"; + + private ClusterHash() { + } + + public static long hashLong(String value) { + return ByteBuffer.wrap(hash(value)).getLong() & Long.MAX_VALUE; + } + + public static String hashHex(String value) { + return HexFormat.of().formatHex(hash(value)); + } + + private static byte[] hash(String value) { + try { + MessageDigest digest = MessageDigest.getInstance(HASH_ALGORITHM); + return digest.digest(value.getBytes(StandardCharsets.UTF_8)); + } catch (NoSuchAlgorithmException exception) { + throw new IllegalStateException(HASH_ALGORITHM + " hash algorithm is not available", exception); + } + } +} diff --git a/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java b/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java index accc031..6c9ddc8 100644 --- a/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java +++ b/src/main/java/org/cache/cluster/hashing/ConsistentHashRing.java @@ -4,10 +4,6 @@ import org.cache.cluster.ClusterInfo; import org.cache.cluster.NodeStatus; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; import java.util.ArrayList; import java.util.HashSet; import java.util.List; @@ -19,8 +15,6 @@ public final class ConsistentHashRing { private static final int DEFAULT_VIRTUAL_NODE_COUNT = 128; - private static final String HASH_ALGORITHM = "SHA-256"; - private final NavigableMap ring = new TreeMap<>(); private final int replicationFactor; @@ -52,7 +46,7 @@ public CacheNode nodeFor(String key) { public List nodesFor(String key) { Objects.requireNonNull(key, "Key must not be null"); - long keyHash = hash(key); + long keyHash = ClusterHash.hashLong(key); List nodes = new ArrayList<>(replicationFactor); Set selectedNodeIds = new HashSet<>(); @@ -70,7 +64,7 @@ private void buildRing(List nodes, int virtualNodeCount) { private void addNode(CacheNode node, int virtualNodeCount) { for (int index = 0; index < virtualNodeCount; index++) { - ring.put(hash(node.getId() + "#" + index), node); + ring.put(ClusterHash.hashLong(node.getId() + "#" + index), node); } } @@ -86,13 +80,4 @@ private void addNodesFromRing(long startHash, List nodes, Set } } - private long hash(String value) { - try { - MessageDigest digest = MessageDigest.getInstance(HASH_ALGORITHM); - byte[] hash = digest.digest(value.getBytes(StandardCharsets.UTF_8)); - return ByteBuffer.wrap(hash).getLong() & Long.MAX_VALUE; - } catch (NoSuchAlgorithmException e) { - throw new IllegalStateException(HASH_ALGORITHM + " hash algorithm is not available", e); - } - } } diff --git a/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java index 71b0ac6..eacffe6 100644 --- a/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java +++ b/src/main/java/org/cache/cluster/routing/ClusterForwardingClient.java @@ -8,9 +8,6 @@ public class ClusterForwardingClient { - private static final List PING_COMMAND = List.of("PING"); - private static final List PONG_RESPONSE = List.of("PONG"); - private final RespCommandClient commandClient; public ClusterForwardingClient() { @@ -28,12 +25,4 @@ public List forward(CacheNode targetNode, List commandParts) { 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) { - return false; - } - } } diff --git a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java index 34bc0e7..333656f 100644 --- a/src/main/java/org/cache/cluster/routing/RoutedCacheService.java +++ b/src/main/java/org/cache/cluster/routing/RoutedCacheService.java @@ -1,7 +1,7 @@ package org.cache.cluster.routing; import org.cache.cluster.CacheNode; -import org.cache.cluster.ClusterInfo; +import org.cache.cluster.ClusterMembership; import org.cache.cluster.hashing.ConsistentHashRing; import org.cache.core.CacheOperations; import org.cache.core.metrics.Snapshot; @@ -16,48 +16,46 @@ public class RoutedCacheService implements CacheOperations { private final CacheOperations localService; private final CacheNode currentNode; private final ClusterForwardingClient forwardingClient; - private final ConsistentHashRing hashRing; private final KeyCodec keyCodec; private final CacheResponseParser responseParser; private final boolean forwardingAllowed; + private final ClusterMembership clusterMembership; public RoutedCacheService( CacheOperations localService, CacheNode currentNode, - ClusterInfo clusterInfo, - ClusterForwardingClient forwardingClient, - KeyCodec keyCodec - ) { - this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, true); - } - - public RoutedCacheService( - CacheOperations localService, - CacheNode currentNode, - ClusterInfo clusterInfo, ClusterForwardingClient forwardingClient, KeyCodec keyCodec, - boolean forwardingAllowed + boolean forwardingAllowed, + ClusterMembership clusterMembership ) { - this(localService, currentNode, clusterInfo, forwardingClient, keyCodec, new CacheResponseParser(), forwardingAllowed); + this( + localService, + currentNode, + forwardingClient, + keyCodec, + new CacheResponseParser(), + forwardingAllowed, + clusterMembership + ); } RoutedCacheService( CacheOperations localService, CacheNode currentNode, - ClusterInfo clusterInfo, ClusterForwardingClient forwardingClient, KeyCodec keyCodec, CacheResponseParser responseParser, - boolean forwardingAllowed + boolean forwardingAllowed, + ClusterMembership clusterMembership ) { this.localService = localService; this.currentNode = currentNode; this.forwardingClient = forwardingClient; - this.hashRing = clusterInfo == null ? null : new ConsistentHashRing(clusterInfo); this.keyCodec = keyCodec; this.responseParser = responseParser; this.forwardingAllowed = forwardingAllowed; + this.clusterMembership = clusterMembership; } @Override @@ -163,6 +161,8 @@ public Snapshot metrics() { } private ReplicationTargets writeTargetsFor(K key) { + ConsistentHashRing hashRing = hashRing(); + if (hashRing == null) { return new ReplicationTargets(true, List.of()); } @@ -189,6 +189,9 @@ private ReplicationTargets writeTargetsFor(K key) { } private List readOwnersFor(K key) { + + ConsistentHashRing hashRing = hashRing(); + if (hashRing == null) { return List.of(currentNode); } @@ -248,6 +251,10 @@ private boolean isCurrentNode(CacheNode owner) { return owner.getId().equals(currentNode.getId()); } + private ConsistentHashRing hashRing() { + return clusterMembership == null ? null : clusterMembership.getHashRing(); + } + private String ownerIds(List owners) { return owners.stream() .map(CacheNode::getId) diff --git a/src/main/java/org/cache/config/CacheConfigLoader.java b/src/main/java/org/cache/config/CacheConfigLoader.java index 94c92d7..85f7df7 100644 --- a/src/main/java/org/cache/config/CacheConfigLoader.java +++ b/src/main/java/org/cache/config/CacheConfigLoader.java @@ -2,6 +2,7 @@ import org.cache.cluster.CacheNode; import org.cache.cluster.ClusterInfo; +import org.cache.cluster.ClusterValidator; import org.cache.cluster.NodeStatus; import org.cache.eviction.EvictionPolicy; import org.cache.eviction.EvictionPolicyType; @@ -15,11 +16,9 @@ import org.springframework.core.io.Resource; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.Properties; -import java.util.Set; public final class CacheConfigLoader { @@ -122,41 +121,11 @@ private ClusterInfo buildClusterInfo(Properties properties) { index++; } - validateClusterInfo(replicationFactor, nodes); + ClusterValidator.validateClusterInfo(replicationFactor, nodes); return new ClusterInfo(replicationFactor, nodes); } - private void validateClusterInfo(int replicationFactor, List nodes) { - if (replicationFactor < 1) { - throw new CacheConfigException("Cluster replication factor must be at least 1"); - } - - if (replicationFactor > nodes.size()) { - throw new CacheConfigException("Cluster replication factor must not exceed number of active nodes"); - } - - Set nodeIds = new HashSet<>(); - Set hostPorts = new HashSet<>(); - - for (CacheNode node : nodes) { - if (!nodeIds.add(node.getId())) { - throw new CacheConfigException("Cluster node ids must be unique: " + node.getId()); - } - - addHostPort(hostPorts, node.getHost(), node.getHttpPort()); - addHostPort(hostPorts, node.getHost(), node.getTcpPort()); - addHostPort(hostPorts, node.getHost(), node.getClusterPort()); - } - } - - private void addHostPort(Set hostPorts, String host, int port) { - String hostPort = host + ":" + port; - if (!hostPorts.add(hostPort)) { - throw new CacheConfigException("Cluster node host-port combinations must be unique: " + hostPort); - } - } - private boolean hasClusterConfig(Properties properties) { return properties.stringPropertyNames() .stream() diff --git a/src/main/java/org/cache/network/http/ClusterController.java b/src/main/java/org/cache/network/http/ClusterController.java new file mode 100644 index 0000000..00aa131 --- /dev/null +++ b/src/main/java/org/cache/network/http/ClusterController.java @@ -0,0 +1,54 @@ +package org.cache.network.http; + +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; +import org.cache.network.http.dto.ClusterNodeRequestDto; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.DeleteMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequestMapping("/cluster") +public class ClusterController { + + private final ClusterMembership clusterMembership; + private final ClusterGossipService clusterGossipService; + + public ClusterController(ClusterMembership clusterMembership, ClusterGossipService clusterGossipService) { + this.clusterMembership = clusterMembership; + this.clusterGossipService = clusterGossipService; + } + + @PostMapping("/nodes") + public ResponseEntity addNode(@RequestBody ClusterNodeRequestDto request) { + try { + clusterMembership.addNode(new CacheNode( + request.id(), + request.host(), + request.httpPort(), + request.tcpPort(), + request.clusterPort() + )); + clusterGossipService.broadcastTopology(); + return ResponseEntity.noContent().build(); + } catch (IllegalArgumentException exception) { + return ResponseEntity.badRequest().build(); + } + } + + @DeleteMapping("/nodes/{nodeId}") + public ResponseEntity removeNode(@PathVariable String nodeId) { + try { + clusterMembership.removeNode(nodeId); + clusterGossipService.broadcastTopology(); + return ResponseEntity.noContent().build(); + } catch (IllegalArgumentException exception) { + return ResponseEntity.badRequest().build(); + } + } +} diff --git a/src/main/java/org/cache/network/http/dto/ClusterNodeRequestDto.java b/src/main/java/org/cache/network/http/dto/ClusterNodeRequestDto.java new file mode 100644 index 0000000..ecfee1a --- /dev/null +++ b/src/main/java/org/cache/network/http/dto/ClusterNodeRequestDto.java @@ -0,0 +1,10 @@ +package org.cache.network.http.dto; + +public record ClusterNodeRequestDto( + String id, + String host, + int httpPort, + int tcpPort, + int clusterPort +) { +} diff --git a/src/main/java/org/cache/network/tcp/connection/RespConnection.java b/src/main/java/org/cache/network/tcp/connection/RespConnection.java index 356b566..39408f0 100644 --- a/src/main/java/org/cache/network/tcp/connection/RespConnection.java +++ b/src/main/java/org/cache/network/tcp/connection/RespConnection.java @@ -27,6 +27,8 @@ import static org.cache.protocol.handlers.ResponseConstants.LIST; import static org.cache.protocol.handlers.ResponseConstants.METRICS; import static org.cache.protocol.handlers.ResponseConstants.SIZE; +import static org.cache.protocol.handlers.ResponseConstants.TOPOLOGY; +import static org.cache.protocol.handlers.ResponseConstants.TOPOLOGY_DIGEST; import static org.cache.protocol.handlers.ResponseConstants.VALUE; public class RespConnection implements ProtocolConnection, AutoCloseable { @@ -186,6 +188,12 @@ private List responseParts(String response) { if (Set.of(ERROR.name(), VALUE.name(), SIZE.name()).contains(prefix)) { return List.of(prefix, parts[1]); } + if (prefix.equals(TOPOLOGY.name())) { + return List.of(prefix, parts[1]); + } + if (prefix.equals(TOPOLOGY_DIGEST.name())) { + return List.of(response.split(WHITESPACE, 3)); + } } if (prefix.equals(LIST.name())) { diff --git a/src/main/java/org/cache/protocol/CommandProcessor.java b/src/main/java/org/cache/protocol/CommandProcessor.java index 7d854e0..7f33e6f 100644 --- a/src/main/java/org/cache/protocol/CommandProcessor.java +++ b/src/main/java/org/cache/protocol/CommandProcessor.java @@ -1,8 +1,13 @@ package org.cache.protocol; -import org.cache.protocol.codec.KeyCodec; +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopologyCodec; import org.cache.core.CacheOperations; +import org.cache.protocol.codec.KeyCodec; import org.cache.protocol.handlers.ClearHandler; +import org.cache.protocol.handlers.ClusterAddNodeHandler; +import org.cache.protocol.handlers.ClusterRemoveNodeHandler; import org.cache.protocol.handlers.CommandHandler; import org.cache.protocol.handlers.CommandType; import org.cache.protocol.handlers.DeleteHandler; @@ -13,6 +18,9 @@ import org.cache.protocol.handlers.PushHandler; import org.cache.protocol.handlers.PutHandler; import org.cache.protocol.handlers.SizeHandler; +import org.cache.protocol.handlers.TopologyApplyHandler; +import org.cache.protocol.handlers.TopologyDigestHandler; +import org.cache.protocol.handlers.TopologyGetHandler; import java.util.EnumMap; import java.util.List; @@ -25,6 +33,29 @@ public class CommandProcessor { private final Map handlers; public CommandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { + this(keyCodec, cacheService, null, null); + } + + public CommandProcessor(KeyCodec keyCodec, CacheOperations cacheService, ClusterMembership clusterMembership) { + this(keyCodec, cacheService, clusterMembership, null); + } + + public CommandProcessor( + KeyCodec keyCodec, + CacheOperations cacheService, + ClusterMembership clusterMembership, + ClusterGossipService clusterGossipService + ) { + this(keyCodec, cacheService, clusterMembership, clusterGossipService, true); + } + + public CommandProcessor( + KeyCodec keyCodec, + CacheOperations cacheService, + ClusterMembership clusterMembership, + ClusterGossipService clusterGossipService, + boolean registerTopologyCommands + ) { this.handlers = new EnumMap<>(CommandType.class); handlers.put(CommandType.PUT, new PutHandler<>(keyCodec, cacheService)); handlers.put(CommandType.GET, new GetHandler<>(keyCodec, cacheService)); @@ -35,6 +66,18 @@ public CommandProcessor(KeyCodec keyCodec, CacheOperations cacheService) { handlers.put(CommandType.PUSH, new PushHandler<>(keyCodec, cacheService)); handlers.put(CommandType.LRANGE, new LrangeHandler<>(keyCodec, cacheService)); handlers.put(CommandType.PING, new PingHandler()); + + if (clusterMembership != null && registerTopologyCommands) { + ClusterTopologyCodec topologyCodec = new ClusterTopologyCodec(); + handlers.put(CommandType.TOPOLOGY_DIGEST, new TopologyDigestHandler(clusterMembership)); + handlers.put(CommandType.TOPOLOGY_GET, new TopologyGetHandler(clusterMembership, topologyCodec)); + handlers.put(CommandType.TOPOLOGY_APPLY, new TopologyApplyHandler(clusterMembership, topologyCodec)); + } + + if (clusterMembership != null && clusterGossipService != null) { + handlers.put(CommandType.CLUSTER_ADD_NODE, new ClusterAddNodeHandler(clusterMembership, clusterGossipService)); + handlers.put(CommandType.CLUSTER_REMOVE_NODE, new ClusterRemoveNodeHandler(clusterMembership, clusterGossipService)); + } } public String process(List commandParts) { diff --git a/src/main/java/org/cache/protocol/handlers/ClusterAddNodeHandler.java b/src/main/java/org/cache/protocol/handlers/ClusterAddNodeHandler.java new file mode 100644 index 0000000..80fffdd --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/ClusterAddNodeHandler.java @@ -0,0 +1,47 @@ +package org.cache.protocol.handlers; + +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterNumberParser; + +import java.util.List; + +public class ClusterAddNodeHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 6; + private static final int NODE_ID_INDEX = 1; + private static final int NODE_HOST_INDEX = 2; + private static final int NODE_HTTP_PORT_INDEX = 3; + private static final int NODE_TCP_PORT_INDEX = 4; + private static final int NODE_CLUSTER_PORT_INDEX = 5; + + private final ClusterMembership clusterMembership; + private final ClusterGossipService clusterGossipService; + + public ClusterAddNodeHandler(ClusterMembership clusterMembership, ClusterGossipService clusterGossipService) { + this.clusterMembership = clusterMembership; + this.clusterGossipService = clusterGossipService; + } + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: CLUSTER_ADD_NODE id host httpPort tcpPort clusterPort"); + } + + try { + clusterMembership.addNode(new CacheNode( + parts.get(NODE_ID_INDEX), + parts.get(NODE_HOST_INDEX), + ClusterNumberParser.parseInt(parts.get(NODE_HTTP_PORT_INDEX), "httpPort"), + ClusterNumberParser.parseInt(parts.get(NODE_TCP_PORT_INDEX), "tcpPort"), + ClusterNumberParser.parseInt(parts.get(NODE_CLUSTER_PORT_INDEX), "clusterPort") + )); + clusterGossipService.broadcastTopology(); + return ResponseConstants.OK.name(); + } catch (IllegalArgumentException exception) { + return TcpResponseSupport.error(exception.getMessage()); + } + } +} diff --git a/src/main/java/org/cache/protocol/handlers/ClusterRemoveNodeHandler.java b/src/main/java/org/cache/protocol/handlers/ClusterRemoveNodeHandler.java new file mode 100644 index 0000000..730b223 --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/ClusterRemoveNodeHandler.java @@ -0,0 +1,35 @@ +package org.cache.protocol.handlers; + +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; + +import java.util.List; + +public class ClusterRemoveNodeHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 2; + private static final int NODE_ID_INDEX = 1; + + private final ClusterMembership clusterMembership; + private final ClusterGossipService clusterGossipService; + + public ClusterRemoveNodeHandler(ClusterMembership clusterMembership, ClusterGossipService clusterGossipService) { + this.clusterMembership = clusterMembership; + this.clusterGossipService = clusterGossipService; + } + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: CLUSTER_REMOVE_NODE id"); + } + + try { + clusterMembership.removeNode(parts.get(NODE_ID_INDEX)); + clusterGossipService.broadcastTopology(); + return ResponseConstants.OK.name(); + } catch (IllegalArgumentException exception) { + return TcpResponseSupport.error(exception.getMessage()); + } + } +} diff --git a/src/main/java/org/cache/protocol/handlers/CommandType.java b/src/main/java/org/cache/protocol/handlers/CommandType.java index c24434b..28549f6 100644 --- a/src/main/java/org/cache/protocol/handlers/CommandType.java +++ b/src/main/java/org/cache/protocol/handlers/CommandType.java @@ -10,5 +10,10 @@ public enum CommandType { PUSH, LRANGE, PING, + TOPOLOGY_DIGEST, + TOPOLOGY_GET, + TOPOLOGY_APPLY, + CLUSTER_ADD_NODE, + CLUSTER_REMOVE_NODE, UNKNOWN; } diff --git a/src/main/java/org/cache/protocol/handlers/ResponseConstants.java b/src/main/java/org/cache/protocol/handlers/ResponseConstants.java index 25f882e..8d4d444 100644 --- a/src/main/java/org/cache/protocol/handlers/ResponseConstants.java +++ b/src/main/java/org/cache/protocol/handlers/ResponseConstants.java @@ -8,5 +8,7 @@ public enum ResponseConstants { VALUE, SIZE, METRICS, - LIST; + LIST, + TOPOLOGY_DIGEST, + TOPOLOGY; } diff --git a/src/main/java/org/cache/protocol/handlers/TopologyApplyHandler.java b/src/main/java/org/cache/protocol/handlers/TopologyApplyHandler.java new file mode 100644 index 0000000..d75ab7b --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/TopologyApplyHandler.java @@ -0,0 +1,36 @@ +package org.cache.protocol.handlers; + +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; +import org.cache.cluster.ClusterTopologyCodec; + +import java.util.List; + +public class TopologyApplyHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 2; + private static final int TOPOLOGY_INDEX = 1; + + private final ClusterMembership clusterMembership; + private final ClusterTopologyCodec topologyCodec; + + public TopologyApplyHandler(ClusterMembership clusterMembership, ClusterTopologyCodec topologyCodec) { + this.clusterMembership = clusterMembership; + this.topologyCodec = topologyCodec; + } + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: TOPOLOGY_APPLY topology"); + } + + try { + ClusterTopology topology = topologyCodec.decode(parts.get(TOPOLOGY_INDEX)); + clusterMembership.applyTopology(topology); + return ResponseConstants.OK.name(); + } catch (IllegalArgumentException exception) { + return TcpResponseSupport.error(exception.getMessage()); + } + } +} diff --git a/src/main/java/org/cache/protocol/handlers/TopologyDigestHandler.java b/src/main/java/org/cache/protocol/handlers/TopologyDigestHandler.java new file mode 100644 index 0000000..ce37083 --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/TopologyDigestHandler.java @@ -0,0 +1,31 @@ +package org.cache.protocol.handlers; + +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; + +import java.util.List; + +public class TopologyDigestHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 1; + + private final ClusterMembership clusterMembership; + + public TopologyDigestHandler(ClusterMembership clusterMembership) { + this.clusterMembership = clusterMembership; + } + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: TOPOLOGY_DIGEST"); + } + + ClusterTopology topology = clusterMembership.currentTopology(); + if (topology == null) { + return TcpResponseSupport.error("cluster is not enabled"); + } + + return "TOPOLOGY_DIGEST " + topology.version() + " " + topology.fingerprint(); + } +} diff --git a/src/main/java/org/cache/protocol/handlers/TopologyGetHandler.java b/src/main/java/org/cache/protocol/handlers/TopologyGetHandler.java new file mode 100644 index 0000000..c9b7e76 --- /dev/null +++ b/src/main/java/org/cache/protocol/handlers/TopologyGetHandler.java @@ -0,0 +1,34 @@ +package org.cache.protocol.handlers; + +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; +import org.cache.cluster.ClusterTopologyCodec; + +import java.util.List; + +public class TopologyGetHandler implements CommandHandler { + + private static final int COMMAND_PARTS_SIZE = 1; + + private final ClusterMembership clusterMembership; + private final ClusterTopologyCodec topologyCodec; + + public TopologyGetHandler(ClusterMembership clusterMembership, ClusterTopologyCodec topologyCodec) { + this.clusterMembership = clusterMembership; + this.topologyCodec = topologyCodec; + } + + @Override + public String handle(List parts) { + if (parts.size() != COMMAND_PARTS_SIZE) { + return TcpResponseSupport.error("usage: TOPOLOGY_GET"); + } + + ClusterTopology topology = clusterMembership.currentTopology(); + if (topology == null) { + return TcpResponseSupport.error("cluster is not enabled"); + } + + return "TOPOLOGY " + topologyCodec.encode(topology); + } +} diff --git a/src/test/java/org/cache/cluster/ClusterGossipServiceTest.java b/src/test/java/org/cache/cluster/ClusterGossipServiceTest.java new file mode 100644 index 0000000..9f074c4 --- /dev/null +++ b/src/test/java/org/cache/cluster/ClusterGossipServiceTest.java @@ -0,0 +1,100 @@ +package org.cache.cluster; + +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Optional; +import java.util.concurrent.ScheduledExecutorService; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ClusterGossipServiceTest { + + private final CacheNode nodeA = node("node-a", 8080, 2020, 10001); + private final CacheNode nodeB = node("node-b", 8081, 2021, 10002); + private final CacheNode nodeC = node("node-c", 8082, 2022, 10003); + private final CacheNode nodeD = node("node-d", 8083, 2023, 10004); + + @Test + void broadcastTopologyPushesCurrentTopologyToPeers() { + ClusterMembership membership = membership(topology(2, nodeA, nodeB)); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterGossipService gossipService = service(membership, membershipClient); + + gossipService.broadcastTopology(); + + verify(membershipClient).applyTopology(nodeB, membership.currentTopology()); + } + + @Test + void gossipPushesLocalTopologyWhenPeerIsOlder() { + ClusterMembership membership = membership(topology(2, nodeA, nodeB)); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterGossipService gossipService = service(membership, membershipClient); + + when(membershipClient.topologyDigest(nodeB)) + .thenReturn(Optional.of(new TopologyDigest(1, "old"))); + + gossipService.gossipOnce(); + + verify(membershipClient).applyTopology(nodeB, membership.currentTopology()); + } + + @Test + void gossipPullsPeerTopologyWhenPeerIsNewer() { + ClusterTopology peerTopology = topology(3, nodeA, nodeB, nodeC); + ClusterMembership membership = membership(topology(2, nodeA, nodeB)); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterGossipService gossipService = service(membership, membershipClient); + + when(membershipClient.topologyDigest(nodeB)) + .thenReturn(Optional.of(new TopologyDigest(3, peerTopology.fingerprint()))); + when(membershipClient.topology(nodeB)).thenReturn(Optional.of(peerTopology)); + + gossipService.gossipOnce(); + + assertEquals(3, membership.currentTopology().version()); + assertEquals(3, membership.currentTopology().nodes().size()); + } + + @Test + void gossipPushesLocalTopologyWhenSameVersionConflictKeepsLocalTopology() { + ClusterTopology localTopology = topology(2, nodeA, nodeB); + ClusterTopology peerTopology = topology(2, nodeA, nodeD); + ClusterMembership membership = membership(localTopology); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterGossipService gossipService = service(membership, membershipClient); + + when(membershipClient.topologyDigest(nodeB)) + .thenReturn(Optional.of(new TopologyDigest(2, peerTopology.fingerprint()))); + when(membershipClient.topology(nodeB)).thenReturn(Optional.of(peerTopology)); + + gossipService.gossipOnce(); + + verify(membershipClient).applyTopology(nodeB, localTopology); + } + + private ClusterGossipService service(ClusterMembership membership, ClusterMembershipClient membershipClient) { + return new ClusterGossipService( + nodeA, + membership, + membershipClient, + mock(ScheduledExecutorService.class) + ); + } + + private ClusterMembership membership(ClusterTopology topology) { + return new ClusterMembership(topology); + } + + private ClusterTopology topology(long version, CacheNode... nodes) { + return new ClusterTopology(version, List.of(nodes), 1, 128); + } + + private CacheNode node(String id, int httpPort, int tcpPort, int clusterPort) { + return new CacheNode(id, "localhost", httpPort, tcpPort, clusterPort); + } +} diff --git a/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java b/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java index 1b3dc05..e737fe3 100644 --- a/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java +++ b/src/test/java/org/cache/cluster/ClusterHealthMonitorTest.java @@ -1,6 +1,5 @@ package org.cache.cluster; -import org.cache.cluster.routing.ClusterForwardingClient; import org.junit.jupiter.api.Test; import java.util.List; @@ -15,14 +14,14 @@ class ClusterHealthMonitorTest { private final CacheNode nodeA = node("node-a", 8080, 2020, 10001); private final CacheNode nodeB = node("node-b", 8081, 2021, 10002); - private final ClusterInfo clusterInfo = new ClusterInfo(1, List.of(nodeA, nodeB)); + private final ClusterMembership clusterMembership = membership(nodeA, nodeB); @Test void checkClusterMovesNodeThroughFailureStatesAfterConsecutiveFailures() { - ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); - ClusterHealthMonitor monitor = monitor(forwardingClient); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterHealthMonitor monitor = monitor(membershipClient); - when(forwardingClient.ping(nodeB)).thenReturn(false); + when(membershipClient.ping(nodeB)).thenReturn(false); monitor.checkCluster(); assertEquals(NodeStatus.HEALTHY, nodeB.getStatus()); @@ -36,10 +35,10 @@ void checkClusterMovesNodeThroughFailureStatesAfterConsecutiveFailures() { @Test void checkClusterRestoresHealthyStatusAfterSuccessfulPing() { - ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); - ClusterHealthMonitor monitor = monitor(forwardingClient); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); + ClusterHealthMonitor monitor = monitor(membershipClient); - when(forwardingClient.ping(nodeB)).thenReturn(false, false, true); + when(membershipClient.ping(nodeB)).thenReturn(false, false, true); monitor.checkCluster(); monitor.checkCluster(); @@ -51,28 +50,32 @@ void checkClusterRestoresHealthyStatusAfterSuccessfulPing() { @Test void checkClusterDoesNothingWhenClusterIsDisabled() { - ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterMembershipClient membershipClient = mock(ClusterMembershipClient.class); ClusterHealthMonitor monitor = new ClusterHealthMonitor( nodeA, - null, - forwardingClient, + new ClusterMembership(null), + membershipClient, mock(ScheduledExecutorService.class) ); monitor.checkCluster(); - verifyNoInteractions(forwardingClient); + verifyNoInteractions(membershipClient); } - private ClusterHealthMonitor monitor(ClusterForwardingClient forwardingClient) { + private ClusterHealthMonitor monitor(ClusterMembershipClient membershipClient) { return new ClusterHealthMonitor( nodeA, - clusterInfo, - forwardingClient, + clusterMembership, + membershipClient, mock(ScheduledExecutorService.class) ); } + private ClusterMembership membership(CacheNode... nodes) { + return new ClusterMembership(new ClusterTopology(0, List.of(nodes), 1, 128)); + } + private CacheNode node(String id, int httpPort, int tcpPort, int clusterPort) { return new CacheNode(id, "localhost", httpPort, tcpPort, clusterPort); } diff --git a/src/test/java/org/cache/cluster/ClusterMembershipClientTest.java b/src/test/java/org/cache/cluster/ClusterMembershipClientTest.java new file mode 100644 index 0000000..bf82b0f --- /dev/null +++ b/src/test/java/org/cache/cluster/ClusterMembershipClientTest.java @@ -0,0 +1,93 @@ +package org.cache.cluster; + +import org.cache.cluster.routing.ClusterForwardingClient; +import org.cache.cluster.routing.ClusterForwardingException; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ClusterMembershipClientTest { + + private final CacheNode node = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + + @Test + void pingReturnsTrueWhenNodeRespondsWithPong() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("PING"))).thenReturn(List.of("PONG")); + + assertTrue(membershipClient.ping(node)); + verify(forwardingClient).forward(node, List.of("PING")); + } + + @Test + void pingReturnsFalseWhenNodeDoesNotRespondWithPong() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("PING"))).thenReturn(List.of("ERROR", "broken")); + + assertFalse(membershipClient.ping(node)); + } + + @Test + void pingReturnsFalseWhenForwardingFails() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("PING"))).thenThrow(new ClusterForwardingException("closed")); + + assertFalse(membershipClient.ping(node)); + } + + @Test + void topologyDigestParsesVersionAndFingerprint() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("TOPOLOGY_DIGEST"))) + .thenReturn(List.of("TOPOLOGY_DIGEST", "3", "abc")); + + var digest = membershipClient.topologyDigest(node).orElseThrow(); + + assertEquals(3, digest.version()); + assertEquals("abc", digest.fingerprint()); + } + + @Test + void topologyParsesFullTopologyResponse() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterTopology topology = new ClusterTopology(3, List.of(node), 1, 128); + String encodedTopology = new ClusterTopologyCodec().encode(topology); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("TOPOLOGY_GET"))) + .thenReturn(List.of("TOPOLOGY", encodedTopology)); + + ClusterTopology response = membershipClient.topology(node).orElseThrow(); + + assertEquals(3, response.version()); + assertEquals(List.of(node), response.nodes()); + } + + @Test + void applyTopologySendsEncodedTopology() { + ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); + ClusterTopology topology = new ClusterTopology(3, List.of(node), 1, 128); + String encodedTopology = new ClusterTopologyCodec().encode(topology); + ClusterMembershipClient membershipClient = new ClusterMembershipClient(forwardingClient); + + when(forwardingClient.forward(node, List.of("TOPOLOGY_APPLY", encodedTopology))).thenReturn(List.of("OK")); + + assertTrue(membershipClient.applyTopology(node, topology)); + verify(forwardingClient).forward(node, List.of("TOPOLOGY_APPLY", encodedTopology)); + } +} diff --git a/src/test/java/org/cache/cluster/ClusterMembershipTest.java b/src/test/java/org/cache/cluster/ClusterMembershipTest.java new file mode 100644 index 0000000..6c7343c --- /dev/null +++ b/src/test/java/org/cache/cluster/ClusterMembershipTest.java @@ -0,0 +1,106 @@ +package org.cache.cluster; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class ClusterMembershipTest { + + private final CacheNode nodeA = node("node-a", 8080, 2020, 10001); + private final CacheNode nodeB = node("node-b", 8081, 2021, 10002); + private final CacheNode nodeC = node("node-c", 8082, 2022, 10003); + private final CacheNode nodeD = node("node-d", "localhost", 8083, 2019, 10004); + private final CacheNode nodeE = node("node-e", "localhost", 8084, 2023, 10005); + + @Test + void addNodeIncrementsVersionAndRebuildsTopology() { + ClusterMembership membership = membership(nodeA, nodeB); + + membership.addNode(nodeC); + + assertEquals(1, membership.currentTopology().version()); + assertEquals(3, membership.currentTopology().nodes().size()); + assertTrue(membership.findNode("node-c").isPresent()); + } + + @Test + void removeNodeIncrementsVersionAndRebuildsTopology() { + ClusterMembership membership = membership(nodeA, nodeB, nodeC); + + membership.removeNode("node-c"); + + assertEquals(1, membership.currentTopology().version()); + assertFalse(membership.findNode("node-c").isPresent()); + } + + @Test + void applyTopologyAcceptsOnlyNewerTopology() { + ClusterMembership membership = membership(nodeA, nodeB); + ClusterTopology olderTopology = new ClusterTopology(-1, List.of(nodeA, nodeB, nodeC), 1, 128); + ClusterTopology newerTopology = new ClusterTopology(2, List.of(nodeA, nodeB, nodeC), 1, 128); + + assertFalse(membership.applyTopology(olderTopology)); + assertEquals(2, membership.currentTopology().nodes().size()); + + assertTrue(membership.applyTopology(newerTopology)); + assertEquals(2, membership.currentTopology().version()); + assertEquals(3, membership.currentTopology().nodes().size()); + } + + @Test + void applyTopologyUsesSmallerTcpAddressesAsSameVersionTieBreaker() { + ClusterMembership membership = membership(nodeA, nodeB); + ClusterTopology sameVersionWithSmallerTcpAddress = new ClusterTopology(0, List.of(nodeA, nodeD), 1, 128); + + assertTrue(membership.applyTopology(sameVersionWithSmallerTcpAddress)); + + assertTrue(membership.findNode("node-d").isPresent()); + assertFalse(membership.findNode("node-b").isPresent()); + } + + @Test + void applyTopologyRejectsSameVersionWithLargerTcpAddresses() { + ClusterMembership membership = membership(nodeA, nodeB); + ClusterTopology sameVersionWithLargerTcpAddress = new ClusterTopology(0, List.of(nodeA, nodeE), 1, 128); + + assertFalse(membership.applyTopology(sameVersionWithLargerTcpAddress)); + + assertTrue(membership.findNode("node-b").isPresent()); + assertFalse(membership.findNode("node-e").isPresent()); + } + + @Test + void markStatusUpdatesCurrentNodeStatus() { + ClusterMembership membership = membership(nodeA, nodeB); + + membership.markStatus("node-b", NodeStatus.UNAVAILABLE); + + assertEquals(NodeStatus.UNAVAILABLE, nodeB.getStatus()); + } + + @Test + void removeNodeRejectsUnknownNode() { + ClusterMembership membership = membership(nodeA, nodeB); + + var exception = assertThrows(CacheInfoException.class, () -> membership.removeNode("node-c")); + + assertEquals("Cluster node does not exist: node-c", exception.getMessage()); + } + + private ClusterMembership membership(CacheNode... nodes) { + return new ClusterMembership(new ClusterTopology(0, List.of(nodes), 1, 128)); + } + + private CacheNode node(String id, int httpPort, int tcpPort, int clusterPort) { + return new CacheNode(id, "localhost", httpPort, tcpPort, clusterPort); + } + + private CacheNode node(String id, String host, int httpPort, int tcpPort, int clusterPort) { + return new CacheNode(id, host, httpPort, tcpPort, clusterPort); + } +} diff --git a/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java index f91e9c6..cc58bd4 100644 --- a/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java +++ b/src/test/java/org/cache/cluster/routing/ClusterForwardingClientTest.java @@ -8,9 +8,7 @@ import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -40,34 +38,4 @@ void forwardWrapsIoFailure() throws Exception { assertThrows(ClusterForwardingException.class, () -> forwardingClient.forward(node, List.of("GET", "fruit"))); } - @Test - void pingReturnsTrueWhenNodeRespondsWithPong() throws Exception { - RespCommandClient commandClient = mock(RespCommandClient.class); - ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); - - when(commandClient.send("localhost", 10001, List.of("PING"))).thenReturn(List.of("PONG")); - - assertTrue(forwardingClient.ping(node)); - verify(commandClient).send("localhost", 10001, List.of("PING")); - } - - @Test - void pingReturnsFalseWhenNodeDoesNotRespondWithPong() throws Exception { - RespCommandClient commandClient = mock(RespCommandClient.class); - ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); - - when(commandClient.send("localhost", 10001, List.of("PING"))).thenReturn(List.of("ERROR", "broken")); - - assertFalse(forwardingClient.ping(node)); - } - - @Test - void pingReturnsFalseWhenConnectionFails() throws Exception { - RespCommandClient commandClient = mock(RespCommandClient.class); - ClusterForwardingClient forwardingClient = new ClusterForwardingClient(commandClient); - - when(commandClient.send("localhost", 10001, List.of("PING"))).thenThrow(new IOException("closed")); - - assertFalse(forwardingClient.ping(node)); - } } diff --git a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java index 64a8319..fb7bc4a 100644 --- a/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java +++ b/src/test/java/org/cache/cluster/routing/RoutedCacheServiceTest.java @@ -2,6 +2,8 @@ import org.cache.cluster.CacheNode; import org.cache.cluster.ClusterInfo; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; import org.cache.cluster.hashing.ConsistentHashRing; import org.cache.core.CacheOperations; import org.cache.protocol.codec.StringKeyCodec; @@ -34,9 +36,10 @@ void getStringUsesLocalServiceWhenCurrentNodeOwnsKey() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - clusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(clusterInfo) ); when(localService.getString(key)).thenReturn(Optional.of("apple")); @@ -54,9 +57,10 @@ void getStringForwardsWhenAnotherNodeOwnsKey() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - clusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(clusterInfo) ); when(forwardingClient.forward(nodeB, List.of("GET", key))).thenReturn(List.of("VALUE", "apple")); @@ -74,9 +78,10 @@ void getStringFallsBackToRemoteReplicaWhenPrimaryForwardFails() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(replicatedClusterInfo) ); when(forwardingClient.forward(nodeB, List.of("GET", key))) @@ -98,9 +103,10 @@ void getStringFallsBackToLocalReplicaWhenPrimaryForwardFails() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(replicatedClusterInfo) ); when(forwardingClient.forward(primaryOwner, List.of("GET", key))) @@ -124,10 +130,10 @@ void getStringRejectsRemoteOwnerWhenForwardingIsDisabled() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, keyCodec, - false + false, + membership(replicatedClusterInfo) ); var exception = assertThrows(ClusterForwardingException.class, () -> service.getString(key)); @@ -149,9 +155,10 @@ void putStringWritesToLocalAndRemoteReplicas() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(replicatedClusterInfo) ); for (CacheNode owner : remoteOwners) { @@ -174,9 +181,10 @@ void putStringWritesToAllRemoteReplicasWhenCurrentNodeIsNotReplicaOwner() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(replicatedClusterInfo) ); when(forwardingClient.forward(nodeB, List.of("PUT", key, "apple", "1000"))).thenReturn(List.of("OK")); @@ -197,10 +205,10 @@ void putStringExecutesLocallyOnReplicaOwnerWhenForwardingIsDisabled() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, keyCodec, - false + false, + membership(replicatedClusterInfo) ); service.putString(key, "apple", 1_000); @@ -221,10 +229,10 @@ void putStringRejectsNonReplicaOwnerWhenForwardingIsDisabled() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, keyCodec, - false + false, + membership(replicatedClusterInfo) ); var exception = assertThrows(ClusterForwardingException.class, () -> service.putString(key, "apple", 1_000)); @@ -242,9 +250,10 @@ void lrangeForwardsAndReturnsListValues() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - clusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(clusterInfo) ); when(forwardingClient.forward(nodeB, List.of("LRANGE", key, "0", "2"))).thenReturn(List.of("one", "two")); @@ -262,9 +271,10 @@ void lrangeFallsBackToRemoteReplicaWhenPrimaryForwardFails() { RoutedCacheService service = new RoutedCacheService<>( localService, nodeA, - replicatedClusterInfo, forwardingClient, - keyCodec + keyCodec, + true, + membership(replicatedClusterInfo) ); when(forwardingClient.forward(nodeB, List.of("LRANGE", key, "0", "2"))) @@ -281,7 +291,14 @@ void lrangeFallsBackToRemoteReplicaWhenPrimaryForwardFails() { void clusterDisabledUsesLocalService() { CacheOperations localService = localService(); ClusterForwardingClient forwardingClient = mock(ClusterForwardingClient.class); - RoutedCacheService service = new RoutedCacheService<>(localService, nodeA, null, forwardingClient, keyCodec); + RoutedCacheService service = new RoutedCacheService<>( + localService, + nodeA, + forwardingClient, + keyCodec, + true, + new ClusterMembership(null) + ); when(localService.getString("fruit")).thenReturn(Optional.of("apple")); @@ -295,6 +312,10 @@ private CacheOperations localService() { return mock(CacheOperations.class); } + private ClusterMembership membership(ClusterInfo clusterInfo) { + return new ClusterMembership(new ClusterTopology(0, clusterInfo.nodes(), clusterInfo.replicationFactor(), 128)); + } + private String keyOwnedBy(CacheNode owner) { ConsistentHashRing ring = new ConsistentHashRing(clusterInfo); diff --git a/src/test/java/org/cache/network/http/ClusterControllerTest.java b/src/test/java/org/cache/network/http/ClusterControllerTest.java new file mode 100644 index 0000000..952fb08 --- /dev/null +++ b/src/test/java/org/cache/network/http/ClusterControllerTest.java @@ -0,0 +1,48 @@ +package org.cache.network.http; + +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; +import org.cache.network.http.dto.ClusterNodeRequestDto; +import org.junit.jupiter.api.Test; +import org.springframework.http.HttpStatus; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +class ClusterControllerTest { + + private final CacheNode nodeA = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + private final CacheNode nodeB = new CacheNode("node-b", "localhost", 8081, 2021, 10002); + + @Test + void addNodeUpdatesMembership() { + ClusterMembership membership = new ClusterMembership(new ClusterTopology(0, List.of(nodeA), 1, 128)); + ClusterGossipService gossipService = mock(ClusterGossipService.class); + ClusterController controller = new ClusterController(membership, gossipService); + + var response = controller.addNode(new ClusterNodeRequestDto("node-b", "localhost", 8081, 2021, 10002)); + + assertEquals(HttpStatus.NO_CONTENT, response.getStatusCode()); + assertTrue(membership.findNode("node-b").isPresent()); + verify(gossipService).broadcastTopology(); + } + + @Test + void removeNodeUpdatesMembership() { + ClusterMembership membership = new ClusterMembership(new ClusterTopology(0, List.of(nodeA, nodeB), 1, 128)); + ClusterGossipService gossipService = mock(ClusterGossipService.class); + ClusterController controller = new ClusterController(membership, gossipService); + + var response = controller.removeNode("node-b"); + + assertEquals(HttpStatus.NO_CONTENT, response.getStatusCode()); + assertTrue(membership.findNode("node-b").isEmpty()); + verify(gossipService).broadcastTopology(); + } +} diff --git a/src/test/java/org/cache/protocol/CommandProcessorTest.java b/src/test/java/org/cache/protocol/CommandProcessorTest.java index 6d626ec..09e215f 100644 --- a/src/test/java/org/cache/protocol/CommandProcessorTest.java +++ b/src/test/java/org/cache/protocol/CommandProcessorTest.java @@ -1,5 +1,10 @@ package org.cache.protocol; +import org.cache.cluster.CacheNode; +import org.cache.cluster.ClusterGossipService; +import org.cache.cluster.ClusterMembership; +import org.cache.cluster.ClusterTopology; +import org.cache.cluster.ClusterTopologyCodec; import org.cache.core.LocalCache; import org.cache.core.ValueType; import org.cache.eviction.LruEvictionPolicy; @@ -14,6 +19,9 @@ import java.util.List; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; class CommandProcessorTest { @@ -128,6 +136,97 @@ void processPingReturnsPongOrEchoValue() { } } + @Test + void processTopologyCommandsWhenMembershipIsConfigured() { + try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { + CacheNode nodeA = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + CacheNode nodeB = new CacheNode("node-b", "localhost", 8081, 2021, 10002); + ClusterMembership membership = new ClusterMembership(new ClusterTopology(1, List.of(nodeA), 1, 128)); + CommandProcessor processor = new CommandProcessor<>( + new StringKeyCodec(), + new CacheService<>(cache, valueCodecs()), + membership + ); + ClusterTopology incomingTopology = new ClusterTopology(2, List.of(nodeA, nodeB), 1, 128); + String encodedTopology = new ClusterTopologyCodec().encode(incomingTopology); + + assertEquals("OK", processor.process(List.of("TOPOLOGY_APPLY", encodedTopology))); + assertEquals(2, membership.currentTopology().version()); + assertEquals( + "TOPOLOGY_DIGEST 2 " + membership.currentTopology().fingerprint(), + processor.process(List.of("TOPOLOGY_DIGEST")) + ); + assertEquals( + "TOPOLOGY " + new ClusterTopologyCodec().encode(membership.currentTopology()), + processor.process(List.of("TOPOLOGY_GET")) + ); + } + } + + @Test + void processClusterAddAndRemoveNodeCommandsWhenGossipIsConfigured() { + try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { + CacheNode nodeA = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + ClusterMembership membership = new ClusterMembership(new ClusterTopology(1, List.of(nodeA), 1, 128)); + ClusterGossipService gossipService = mock(ClusterGossipService.class); + CommandProcessor processor = new CommandProcessor<>( + new StringKeyCodec(), + new CacheService<>(cache, valueCodecs()), + membership, + gossipService + ); + + assertEquals( + "OK", + processor.process(List.of("CLUSTER_ADD_NODE", "node-b", "localhost", "8081", "2021", "10002")) + ); + assertEquals(2, membership.currentTopology().version()); + assertEquals(2, membership.currentTopology().nodes().size()); + verify(gossipService).broadcastTopology(); + + assertEquals("OK", processor.process(List.of("CLUSTER_REMOVE_NODE", "node-b"))); + assertEquals(3, membership.currentTopology().version()); + assertEquals(1, membership.currentTopology().nodes().size()); + verify(gossipService, times(2)).broadcastTopology(); + } + } + + @Test + void processClusterAddNodeIsUnknownWhenGossipIsNotConfigured() { + try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { + CommandProcessor processor = processor(cache); + + assertEquals( + "ERROR unknown command", + processor.process(List.of("CLUSTER_ADD_NODE", "node-b", "localhost", "8081", "2021", "10002")) + ); + } + } + + @Test + void processClusterAddAndRemoveNodeCommandsWithoutTopologyCommands() { + try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) { + CacheNode nodeA = new CacheNode("node-a", "localhost", 8080, 2020, 10001); + ClusterMembership membership = new ClusterMembership(new ClusterTopology(1, List.of(nodeA), 1, 128)); + ClusterGossipService gossipService = mock(ClusterGossipService.class); + CommandProcessor processor = new CommandProcessor<>( + new StringKeyCodec(), + new CacheService<>(cache, valueCodecs()), + membership, + gossipService, + false + ); + + assertEquals( + "OK", + processor.process(List.of("CLUSTER_ADD_NODE", "node-b", "localhost", "8081", "2021", "10002")) + ); + assertEquals("OK", processor.process(List.of("CLUSTER_REMOVE_NODE", "node-b"))); + assertEquals("ERROR unknown command", processor.process(List.of("TOPOLOGY_GET"))); + verify(gossipService, times(2)).broadcastTopology(); + } + } + @Test void processReturnsInvalidUsageErrors() { try (var cache = new LocalCache(10, new LruEvictionPolicy<>())) {