From b8c54362931b817a84c91f8d758aa63995ecb4a1 Mon Sep 17 00:00:00 2001 From: shunsaker Date: Wed, 10 Jul 2024 02:03:35 +0200 Subject: [PATCH] Add configurable batchlog endpoint strategies Batchlog endpoint strategy was previously only random placements on other racks. Options now are random_remote, prefer_local, dynamic_remote, and dynamic. patch by Shayne Hunsaker; reviewed by Mick Semb Wever, Brandon Williams for CASSANDRA-18120 --- CHANGES.txt | 1 + conf/cassandra.yaml | 26 + .../config/CassandraRelevantProperties.java | 4 + .../org/apache/cassandra/config/Config.java | 63 ++ .../cassandra/config/DatabaseDescriptor.java | 16 + .../locator/DynamicEndpointSnitch.java | 5 +- .../cassandra/locator/ReplicaPlans.java | 200 +++- .../cassandra/service/StorageService.java | 10 + .../service/StorageServiceMBean.java | 7 + .../batchlog/BatchlogEndpointFilterTest.java | 923 +++++++++++++++++- .../config/DatabaseDescriptorRefTest.java | 1 + 11 files changed, 1188 insertions(+), 68 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 934b054aac..486d8ad7d2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.15 + * Add configurable batchlog endpoint strategies: random_remote, prefer_local, dynamic_remote, and dynamic (CASSANDRA-18120) * Fix bash-completion for debian distro (CASSANDRA-19999) * Ensure thread-safety for CommitLogArchiver in CommitLog (CASSANDRA-19960) * Fix text containing "/*" being interpreted as multiline comment in cqlsh (CASSANDRA-17667) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index b5e6af8767..835a8f4fa6 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -103,6 +103,32 @@ max_hints_file_size_in_mb: 128 # reduced proportionally to the number of nodes in the cluster. batchlog_replay_throttle_in_kb: 1024 +# Strategy to choose the batchlog storage endpoints. +# +# Available options: +# +# - random_remote +# Default, purely random, prevents the local rack, if possible. +# +# - prefer_local +# Similar to random_remote. Random, except that one of the replications will go to the local rack, +# which mean it offers lower availability guarantee than random_remote or dynamic_remote. +# +# - dynamic_remote +# Using DynamicEndpointSnitch to select batchlog storage endpoints, prevents the +# local rack, if possible. This strategy offers the same availability guarantees +# as random_remote but selects the fastest endpoints according to the DynamicEndpointSnitch. +# (DynamicEndpointSnitch currently only tracks reads and not writes - i.e. write-only +# (or mostly-write) workloads might not benefit from this strategy.) +# Note: this strategy will fall back to random_remote, if dynamic_snitch is not enabled. +# +# - dynamic +# Mostly the same as dynamic_remote, except that local rack is not excluded, which mean it offers lower +# availability guarantee than random_remote or dynamic_remote. +# Note: this strategy will fall back to random_remote, if dynamic_snitch is not enabled. +# +# batchlog_endpoint_strategy: random_remote + # Authentication backend, implementing IAuthenticator; used to identify users # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllAuthenticator, # PasswordAuthenticator}. diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 0377bc40e2..1757768897 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -256,6 +256,10 @@ public enum CassandraRelevantProperties */ GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED("cassandra.gossip_settle_poll_success_required", "3"), + /** + * Number of replicas required to store batchlog for atomicity, only accepts values of 1 or 2. + */ + REQUIRED_BATCHLOG_REPLICA_COUNT("cassandra.batchlog.required_replica_count", "2") ; CassandraRelevantProperties(String key, String defaultVal) diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index a9cdca46a1..995aaa69d2 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -292,6 +292,7 @@ public class Config public int hinted_handoff_throttle_in_kb = 1024; public int batchlog_replay_throttle_in_kb = 1024; + public BatchlogEndpointStrategy batchlog_endpoint_strategy = BatchlogEndpointStrategy.random_remote; public int max_hints_delivery_threads = 2; public int hints_flush_period_in_ms = 10000; public int max_hints_file_size_in_mb = 128; @@ -681,6 +682,68 @@ public class Config exception } + public enum BatchlogEndpointStrategy + { + /** + * Old, conventional strategy to select batchlog storage endpoints. + * Purely random, prevents the local rack, if possible. + */ + random_remote(false, false), + + /** + * Random, except that one of the replications will go to the local rack. + * Which means this strategy offers lower availability guarantees than + * {@link #random_remote} or {@link #dynamic_remote}. + */ + prefer_local(false, true), + + /** + * Strategy using {@link Config#dynamic_snitch} ({@link org.apache.cassandra.locator.DynamicEndpointSnitch}) + * to select batchlog storage endpoints. Prevents the local rack, if possible. + * + * This strategy offers the same availability guarantees as {@link #random_remote} but selects the + * fastest endpoints according to the {@link org.apache.cassandra.locator.DynamicEndpointSnitch}. + * + * Hint: {@link org.apache.cassandra.locator.DynamicEndpointSnitch} tracks reads and not writes - i.e. + * write-only (or mostly-write) workloads might not benefit from this strategy. + * + * Note: this strategy will fall back to {@link #random_remote}, if {@link #dynamic_snitch} is not enabled. + */ + dynamic_remote(true, false), + + /** + * Strategy using {@link Config#dynamic_snitch} ({@link org.apache.cassandra.locator.DynamicEndpointSnitch}) + * to select batchlog storage endpoints. Does not prevent the local rack. + * + * Since the local rack is not excluded, this strategy offers lower availability guarantees than + * {@link #random_remote} or {@link #dynamic_remote}. + * + * Hint: {@link org.apache.cassandra.locator.DynamicEndpointSnitch} tracks reads and not writes - i.e. + * write-only (or mostly-write) workloads might not benefit from this strategy. + * + * Note: this strategy will fall back to {@link #random_remote}, if {@link #dynamic_snitch} is not enabled. + */ + dynamic(true, true); + + /** + * If true, dynamic snitch response times will be used to select more responsive nodes to write the batchlog to. + * If false, nodes will be randomly selected. + */ + public final boolean useDynamicSnitchScores; + + /** + * If true, one of the selected nodes will come from the local rack. + * If false, the local rack will not be used except as a last resort with no other racks available. + */ + public final boolean preferLocalRack; + + BatchlogEndpointStrategy(boolean useDynamicSnitchScores, boolean preferLocalRack) + { + this.useDynamicSnitchScores = useDynamicSnitchScores; + this.preferLocalRack = preferLocalRack; + } + } + private static final List SENSITIVE_KEYS = new ArrayList() {{ add("client_encryption_options"); add("server_encryption_options"); diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index b7d47ec9a5..d425e53807 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -2571,6 +2571,22 @@ public class DatabaseDescriptor conf.batchlog_replay_throttle_in_kb = throttleInKB; } + public static boolean isDynamicEndpointSnitch() + { + // not using config.dynamic_snitch because snitch can be changed via JMX + return snitch instanceof DynamicEndpointSnitch; + } + + public static Config.BatchlogEndpointStrategy getBatchlogEndpointStrategy() + { + return conf.batchlog_endpoint_strategy; + } + + public static void setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy batchlogEndpointStrategy) + { + conf.batchlog_endpoint_strategy = batchlogEndpointStrategy; + } + public static int getMaxHintsDeliveryThreads() { return conf.max_hints_delivery_threads; diff --git a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java index 976e1dab22..394afddbfc 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -27,8 +27,8 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import com.codahale.metrics.ExponentiallyDecayingReservoir; - import com.codahale.metrics.Snapshot; + import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.gms.ApplicationState; @@ -41,6 +41,7 @@ import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.MBeanWrapper; + /** * A dynamic snitch that sorts endpoints by latency with an adapted phi failure detector */ @@ -269,7 +270,7 @@ public class DynamicEndpointSnitch extends AbstractEndpointSnitch implements Lat sample.update(unit.toMillis(latency)); } - private void updateScores() // this is expensive + public void updateScores() // this is expensive { if (!StorageService.instance.isGossipActive()) return; diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index 7e7a6a1e33..2d7a93ac76 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -18,16 +18,35 @@ package org.apache.cassandra.locator; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Predicate; + import com.carrotsearch.hppc.ObjectIntHashMap; import com.carrotsearch.hppc.cursors.ObjectObjectCursor; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.HashMultimap; +import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; import com.google.common.collect.ListMultimap; import com.google.common.collect.Lists; import com.google.common.collect.Multimap; +import com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.DecoratedKey; @@ -41,20 +60,7 @@ import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.reads.AlwaysSpeculativeRetryPolicy; import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; - import org.apache.cassandra.utils.FBUtilities; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ThreadLocalRandom; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Predicate; import static com.google.common.collect.Iterables.any; import static com.google.common.collect.Iterables.filter; @@ -71,6 +77,16 @@ public class ReplicaPlans { private static final Logger logger = LoggerFactory.getLogger(ReplicaPlans.class); + private static final int REQUIRED_BATCHLOG_REPLICA_COUNT + = Math.max(1, Math.min(2, CassandraRelevantProperties.REQUIRED_BATCHLOG_REPLICA_COUNT.getInt())); + + static + { + int batchlogReplicaCount = CassandraRelevantProperties.REQUIRED_BATCHLOG_REPLICA_COUNT.getInt(); + if (batchlogReplicaCount < 1 || 2 < batchlogReplicaCount) + logger.warn("System property {} was set to {} but must be 1 or 2. Running with {}", CassandraRelevantProperties.REQUIRED_BATCHLOG_REPLICA_COUNT.getKey(), batchlogReplicaCount, REQUIRED_BATCHLOG_REPLICA_COUNT); + } + public static boolean isSufficientLiveReplicasForRead(AbstractReplicationStrategy replicationStrategy, ConsistencyLevel consistencyLevel, Endpoints liveReplicas) { switch (consistencyLevel) @@ -225,10 +241,22 @@ public class ReplicaPlans // - replicas should be in the local datacenter // - choose min(2, number of qualifying candiates above) // - allow the local node to be the only replica only if it's a single-node DC - Collection chosenEndpoints = filterBatchlogEndpoints(snitch.getLocalRack(), localEndpoints); + Collection chosenEndpoints = filterBatchlogEndpoints(false, snitch.getLocalRack(), localEndpoints); - if (chosenEndpoints.isEmpty() && isAny) - chosenEndpoints = Collections.singleton(FBUtilities.getBroadcastAddressAndPort()); + // Batchlog is hosted by either one node or two nodes from different racks. + ConsistencyLevel consistencyLevel = chosenEndpoints.size() == 1 ? ConsistencyLevel.ONE : ConsistencyLevel.TWO; + + if (chosenEndpoints.isEmpty()) + { + if (isAny) + chosenEndpoints = Collections.singleton(FBUtilities.getBroadcastAddressAndPort()); + else + // UnavailableException instead of letting the batchlog write unnecessarily timeout + throw new UnavailableException("Cannot achieve consistency level " + consistencyLevel + + " for batchlog in local DC, required:" + REQUIRED_BATCHLOG_REPLICA_COUNT + + ", available:" + 0, + consistencyLevel, REQUIRED_BATCHLOG_REPLICA_COUNT, 0); + } Keyspace systemKeypsace = Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME); ReplicaLayout.ForTokenWrite liveAndDown = ReplicaLayout.forTokenWrite( @@ -236,36 +264,34 @@ public class ReplicaPlans SystemReplicas.getSystemReplicas(chosenEndpoints).forToken(token), EndpointsForToken.empty(token) ); - // Batchlog is hosted by either one node or two nodes from different racks. - ConsistencyLevel consistencyLevel = liveAndDown.all().size() == 1 ? ConsistencyLevel.ONE : ConsistencyLevel.TWO; // assume that we have already been given live endpoints, and skip applying the failure detector return forWrite(systemKeypsace, consistencyLevel, liveAndDown, liveAndDown, writeAll); } - private static Collection filterBatchlogEndpoints(String localRack, - Multimap endpoints) + @VisibleForTesting + public static Collection filterBatchlogEndpoints(boolean preferLocalRack, String localRack, + Multimap endpoints) { - return filterBatchlogEndpoints(localRack, - endpoints, - Collections::shuffle, - FailureDetector.isEndpointAlive, - ThreadLocalRandom.current()::nextInt); + return DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores && DatabaseDescriptor.isDynamicEndpointSnitch() + ? filterBatchlogEndpointsDynamic(preferLocalRack,localRack, endpoints, FailureDetector.isEndpointAlive) + : filterBatchlogEndpointsRandom(preferLocalRack, localRack, endpoints, + Collections::shuffle, + FailureDetector.isEndpointAlive, + ThreadLocalRandom.current()::nextInt); } - // Collect a list of candidates for batchlog hosting. If possible these will be two nodes from different racks. - @VisibleForTesting - public static Collection filterBatchlogEndpoints(String localRack, - Multimap endpoints, - Consumer> shuffle, - Predicate isAlive, - Function indexPicker) + private static ListMultimap validate(boolean preferLocalRack, String localRack, + Multimap endpoints, + Predicate isAlive) { + int endpointCount = endpoints.values().size(); // special case for single-node data centers - if (endpoints.values().size() == 1) - return endpoints.values(); + if (endpointCount <= REQUIRED_BATCHLOG_REPLICA_COUNT) + return ArrayListMultimap.create(endpoints); // strip out dead endpoints and localhost - ListMultimap validated = ArrayListMultimap.create(); + int rackCount = endpoints.keySet().size(); + ListMultimap validated = ArrayListMultimap.create(rackCount, endpointCount / rackCount); for (Map.Entry entry : endpoints.entries()) { InetAddressAndPort addr = entry.getValue(); @@ -273,15 +299,45 @@ public class ReplicaPlans validated.put(entry.getKey(), entry.getValue()); } - if (validated.size() <= 2) - return validated.values(); + // return early if no more than 2 nodes: + if (validated.size() <= REQUIRED_BATCHLOG_REPLICA_COUNT) + return validated; - if (validated.size() - validated.get(localRack).size() >= 2) + // if the local rack is not preferred and there are enough nodes in other racks, remove it: + if (!(DatabaseDescriptor.getBatchlogEndpointStrategy().preferLocalRack || preferLocalRack) + && validated.size() - validated.get(localRack).size() >= REQUIRED_BATCHLOG_REPLICA_COUNT) { // we have enough endpoints in other racks validated.removeAll(localRack); } + return validated; + } + + // Collect a list of candidates for batchlog hosting. If possible these will be two nodes from different racks. + // Replicas are picked manually: + // - replicas should be alive according to the failure detector + // - replicas should be in the local datacenter + // - choose min(2, number of qualifying candiates above) + // - allow the local node to be the only replica only if it's a single-node DC + @VisibleForTesting + public static Collection filterBatchlogEndpointsRandom(boolean preferLocalRack, String localRack, + Multimap endpoints, + Consumer> shuffle, + Predicate isAlive, + Function indexPicker) + { + ListMultimap validated = validate(preferLocalRack, localRack, endpoints, isAlive); + + // return early if no more than 2 nodes: + if (validated.size() <= REQUIRED_BATCHLOG_REPLICA_COUNT) + return validated.values(); + + /* + * if we have only 1 `other` rack to select replicas from (whether it be the local rack or a single non-local rack), + * pick two random nodes from there and return early; + * we are guaranteed to have at least two nodes in the single remaining rack because of the above if block. + */ if (validated.keySet().size() == 1) { /* @@ -291,24 +347,32 @@ public class ReplicaPlans */ List otherRack = Lists.newArrayList(validated.values()); shuffle.accept(otherRack); - return otherRack.subList(0, 2); + return otherRack.subList(0, REQUIRED_BATCHLOG_REPLICA_COUNT); } // randomize which racks we pick from if more than 2 remaining Collection racks; - if (validated.keySet().size() == 2) + if (validated.keySet().size() == REQUIRED_BATCHLOG_REPLICA_COUNT) { racks = validated.keySet(); } + else if (preferLocalRack || DatabaseDescriptor.getBatchlogEndpointStrategy().preferLocalRack) + { + List nonLocalRacks = Lists.newArrayList(Sets.difference(validated.keySet(), ImmutableSet.of(localRack))); + racks = new LinkedHashSet<>(); + racks.add(localRack); + racks.add(nonLocalRacks.get(indexPicker.apply(nonLocalRacks.size()))); + } else { racks = Lists.newArrayList(validated.keySet()); shuffle.accept((List) racks); } - // grab a random member of up to two racks - List result = new ArrayList<>(2); - for (String rack : Iterables.limit(racks, 2)) + // grab two random nodes from two different racks + + List result = new ArrayList<>(REQUIRED_BATCHLOG_REPLICA_COUNT); + for (String rack : Iterables.limit(racks, REQUIRED_BATCHLOG_REPLICA_COUNT)) { List rackMembers = validated.get(rack); result.add(rackMembers.get(indexPicker.apply(rackMembers.size()))); @@ -317,6 +381,56 @@ public class ReplicaPlans return result; } + @VisibleForTesting + public static Collection filterBatchlogEndpointsDynamic(boolean preferLocalRack, String localRack, + Multimap endpoints, + Predicate isAlive) + { + ListMultimap validated = validate(preferLocalRack, localRack, endpoints, isAlive); + + // return early if no more than 2 nodes: + if (validated.size() <= REQUIRED_BATCHLOG_REPLICA_COUNT) + return validated.values(); + + // sort _all_ nodes to pick the best racks + List sorted = sortByProximity(validated.values()); + + List result = new ArrayList<>(REQUIRED_BATCHLOG_REPLICA_COUNT); + Set racks = new HashSet<>(); + + while (result.size() < REQUIRED_BATCHLOG_REPLICA_COUNT) + { + for (InetAddressAndPort endpoint : sorted) + { + if (result.size() == REQUIRED_BATCHLOG_REPLICA_COUNT) + break; + + if (racks.isEmpty()) + racks.addAll(validated.keySet()); + + String rack = DatabaseDescriptor.getEndpointSnitch().getRack(endpoint); + if (!racks.remove(rack)) + continue; + if (result.contains(endpoint)) + continue; + + result.add(endpoint); + } + } + + return result; + } + + @VisibleForTesting + public static List sortByProximity(Collection endpoints) + { + EndpointsForRange endpointsForRange = SystemReplicas.getSystemReplicas(endpoints); + return DatabaseDescriptor.getEndpointSnitch() + .sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), endpointsForRange) + .endpointList(); + } + + public static ReplicaPlan.ForTokenWrite forReadRepair(Token token, ReplicaPlan.ForRead readPlan) throws UnavailableException { return forWrite(readPlan.keyspace, readPlan.consistencyLevel, token, writeReadRepair(readPlan)); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 221b9c62ae..5b29e4bee9 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -5617,6 +5617,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE updateTopology(); } + public String getBatchlogEndpointStrategy() + { + return DatabaseDescriptor.getBatchlogEndpointStrategy().name(); + } + + public void setBatchlogEndpointStrategy(String batchlogEndpointStrategy) + { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.valueOf(batchlogEndpointStrategy)); + } + /** * Send data to the endpoints that will be responsible for it in the future * diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 34b7fb7ff3..0cc6dbd21a 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -536,6 +536,13 @@ public interface StorageServiceMBean extends NotificationEmitter */ public int getDynamicUpdateInterval(); + public String getBatchlogEndpointStrategy(); + + /** + * See {@link org.apache.cassandra.config.Config.BatchlogEndpointStrategy} for valid values. + */ + public void setBatchlogEndpointStrategy(String batchlogEndpointStrategy); + // allows a user to forcibly 'kill' a sick node public void stopGossiping(); diff --git a/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java b/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java index 8057e997b6..ab749b1dc1 100644 --- a/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java +++ b/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java @@ -18,27 +18,85 @@ package org.apache.cassandra.batchlog; import java.net.UnknownHostException; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.function.Supplier; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import com.google.common.collect.ArrayListMultimap; import com.google.common.collect.ImmutableMultimap; +import com.google.common.collect.Iterables; import com.google.common.collect.Multimap; +import com.google.common.collect.Multimaps; import org.junit.Test; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.locator.DynamicEndpointSnitch; +import org.apache.cassandra.locator.GossipingPropertyFileSnitch; +import org.apache.cassandra.locator.IEndpointSnitch; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.ReplicaPlans; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; -import static org.hamcrest.CoreMatchers.is; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThat; +import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertTrue; public class BatchlogEndpointFilterTest { private static final String LOCAL = "local"; - @Test - public void shouldSelect2HostsFromNonLocalRacks() throws UnknownHostException + // Repeat all tests some more times since we're dealing with random stuff - i.e. increase the + // chance to hit issues. + private static final int repetitions = 100; + private static final InetAddressAndPort[] INET_ADDRESSES = new InetAddressAndPort[0]; + + private DynamicEndpointSnitch dsnitch; + + @BeforeClass + public static void beforeClass() { + DatabaseDescriptor.daemonInitialization(); + DatabaseDescriptor.setBroadcastAddress(endpointAddress(0, 0).address); + } + + @Test + public void shouldUseLocalRackIfPreferLocalParameter() throws UnknownHostException + { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); + Multimap endpoints = ImmutableMultimap. builder() + .put(LOCAL, InetAddressAndPort.getByName("0")) + .put(LOCAL, InetAddressAndPort.getByName("00")) + .put("1", InetAddressAndPort.getByName("1")) + .put("1", InetAddressAndPort.getByName("11")) + .put("2", InetAddressAndPort.getByName("2")) + .put("2", InetAddressAndPort.getByName("22")) + .build(); + Collection result = filterBatchlogEndpointsRandomForTests(true, endpoints); + assertThat(result.size()).isEqualTo(2); + assertThat(result).containsAnyElementsOf(endpoints.get(LOCAL)); + assertThat(result).containsAnyElementsOf(Iterables.concat(endpoints.get("1"), endpoints.get("2"))); + } + + @Test + public void shouldUseLocalRackIfPreferLocalStrategy() throws UnknownHostException + { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.prefer_local); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("0")) .put(LOCAL, InetAddressAndPort.getByName("00")) @@ -47,8 +105,27 @@ public class BatchlogEndpointFilterTest .put("2", InetAddressAndPort.getByName("2")) .put("2", InetAddressAndPort.getByName("22")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); - assertThat(result.size(), is(2)); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(2); + assertThat(result).containsAnyElementsOf(endpoints.get(LOCAL)); + assertThat(result).containsAnyElementsOf(Iterables.concat(endpoints.get("1"), endpoints.get("2"))); + } + + @Test + public void shouldSelect2HostsFromNonLocalRacks() throws UnknownHostException + { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); + Multimap endpoints = ImmutableMultimap. builder() + .put(LOCAL, InetAddressAndPort.getByName("0")) + .put(LOCAL, InetAddressAndPort.getByName("00")) + .put("1", InetAddressAndPort.getByName("1")) + .put("1", InetAddressAndPort.getByName("11")) + .put("2", InetAddressAndPort.getByName("2")) + .put("2", InetAddressAndPort.getByName("22")) + .build(); + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(2); assertTrue(result.contains(InetAddressAndPort.getByName("11"))); assertTrue(result.contains(InetAddressAndPort.getByName("22"))); } @@ -56,6 +133,7 @@ public class BatchlogEndpointFilterTest @Test public void shouldSelectLastHostsFromLastNonLocalRacks() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("00")) .put("1", InetAddressAndPort.getByName("11")) @@ -65,8 +143,9 @@ public class BatchlogEndpointFilterTest .put("3", InetAddressAndPort.getByName("33")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); - assertThat(result.size(), is(2)); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(2); // result should be the last replicas of the last two racks // (Collections.shuffle has been replaced with Collections.reverse for testing) @@ -77,13 +156,15 @@ public class BatchlogEndpointFilterTest @Test public void shouldSelectHostFromLocal() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("0")) .put(LOCAL, InetAddressAndPort.getByName("00")) .put("1", InetAddressAndPort.getByName("1")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); - assertThat(result.size(), is(2)); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(2); assertTrue(result.contains(InetAddressAndPort.getByName("1"))); assertTrue(result.contains(InetAddressAndPort.getByName("0"))); } @@ -91,17 +172,20 @@ public class BatchlogEndpointFilterTest @Test public void shouldReturnPassedEndpointForSingleNodeDC() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("0")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); - assertThat(result.size(), is(1)); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(1); assertTrue(result.contains(InetAddressAndPort.getByName("0"))); } @Test public void shouldSelectTwoRandomHostsFromSingleOtherRack() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("0")) .put(LOCAL, InetAddressAndPort.getByName("00")) @@ -109,7 +193,8 @@ public class BatchlogEndpointFilterTest .put("1", InetAddressAndPort.getByName("11")) .put("1", InetAddressAndPort.getByName("111")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); // result should be the last two non-local replicas // (Collections.shuffle has been replaced with Collections.reverse for testing) assertThat(result.size(), is(2)); @@ -120,13 +205,15 @@ public class BatchlogEndpointFilterTest @Test public void shouldSelectTwoRandomHostsFromSingleRack() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("1")) .put(LOCAL, InetAddressAndPort.getByName("11")) .put(LOCAL, InetAddressAndPort.getByName("111")) .put(LOCAL, InetAddressAndPort.getByName("1111")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); // result should be the last two non-local replicas // (Collections.shuffle has been replaced with Collections.reverse for testing) assertThat(result.size(), is(2)); @@ -137,24 +224,814 @@ public class BatchlogEndpointFilterTest @Test public void shouldSelectOnlyTwoHostsEvenIfLocal() throws UnknownHostException { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.random_remote); Multimap endpoints = ImmutableMultimap. builder() .put(LOCAL, InetAddressAndPort.getByName("1")) .put(LOCAL, InetAddressAndPort.getByName("11")) .build(); - Collection result = filterBatchlogEndpoints(endpoints); - assertThat(result.size(), is(2)); + + Collection result = filterBatchlogEndpointsRandomForTests(false, endpoints); + assertThat(result.size()).isEqualTo(2); assertTrue(result.contains(InetAddressAndPort.getByName("1"))); assertTrue(result.contains(InetAddressAndPort.getByName("11"))); } - private Collection filterBatchlogEndpoints(Multimap endpoints) + private Collection filterBatchlogEndpointsRandomForTests(boolean preferLocalRack, Multimap endpoints) { - return ReplicaPlans.filterBatchlogEndpoints(LOCAL, endpoints, - // Reverse instead of shuffle - Collections::reverse, - // Always alive - (addr) -> true, - // Always pick the last - (size) -> size - 1); + return ReplicaPlans.filterBatchlogEndpointsRandom(preferLocalRack, LOCAL, endpoints, + // Reverse instead of shuffle + Collections::reverse, + // Always alive + (addr) -> true, + // Always pick the last + (size) -> size - 1); } + + private Collection filterBatchlogEndpointsForTests(Multimap endpoints) + { + return DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores ? + filterBatchlogEndpointsDynamicForTests(endpoints) : + filterBatchlogEndpointsRandomForTests(false, endpoints); + } + + private Collection filterBatchlogEndpointsDynamicForTests(Multimap endpoints) { + return ReplicaPlans.filterBatchlogEndpointsDynamic(false, LOCAL, endpoints, x -> true); + } + + + @Test + public void shouldUseCoordinatorForSingleNodeDC() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 1), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 1) + ), this::shouldUseCoordinatorForSingleNodeDC); + } + + private void shouldUseCoordinatorForSingleNodeDC(Multimap endpoints) + { + Collection result = filterBatchlogEndpointsForTests(endpoints); + assertThat(result.size(), is(1)); + assertThat(result, hasItem(endpointAddress(0, 0))); + } + + @Test + public void shouldUseCoordinatorAndTheOtherForTwoNodesInOneRack() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 2), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 2), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 2), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 2), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 2), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 2) + ), this::shouldUseCoordinatorAndTheOtherForTwoNodesInOneRack); + } + + private void shouldUseCoordinatorAndTheOtherForTwoNodesInOneRack(Multimap endpoints) + { + Collection result = filterBatchlogEndpointsForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(new HashSet<>(result).size(), is(2)); + assertThat(result, hasItem(endpointAddress(0, 0))); + assertThat(result, hasItem(endpointAddress(0, 1))); + } + + @Test + public void shouldUseCoordinatorAndTheOtherForTwoNodesInTwoRacks() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 1, 1) + ), this::shouldUseCoordinatorAndTheOtherForTwoNodesInTwoRacks); + } + + private void shouldUseCoordinatorAndTheOtherForTwoNodesInTwoRacks(Multimap endpoints) + { + Collection result = filterBatchlogEndpointsForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(new HashSet<>(result).size(), is(2)); + assertThat(result, hasItem(endpointAddress(0, 0))); + assertThat(result, hasItem(endpointAddress(1, 0))); + } + + @Test + public void shouldSelectOneNodeFromLocalRackAndOneNodeFromTheOtherRack() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 2, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 1, 1) + ), this::shouldSelectOneNodeFromLocalRackAndOneNodeFromTheOtherRack); + } + + private void shouldSelectOneNodeFromLocalRackAndOneNodeFromTheOtherRack(Multimap endpoints) + { + for (int i = 0; i < repetitions; i++) + { + Collection result = filterBatchlogEndpointsForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(new HashSet<>(result).size(), is(2)); + assertThat(result, hasItem(endpointAddress(1, 0))); + assertThat(result, either(hasItem(endpointAddress(0, 0))) + .or(hasItem(endpointAddress(0, 1)))); + } + } + + @Test + public void shouldReturnNoBatchlogEnpointsIfAllAreUnavailable() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 3), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15) + ), this::shouldReturnNoBatchlogEnpointsIfAllAreUnavailable); + } + + private void shouldReturnNoBatchlogEnpointsIfAllAreUnavailable(Multimap endpoints) + { + Predicate isAlive = x -> x.equals(endpointAddress(0, 0)); + for (int i = 0; i < repetitions; i++) + { + Collection result = DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores ? + ReplicaPlans.filterBatchlogEndpointsDynamic(false, LOCAL, endpoints, isAlive) : + ReplicaPlans.filterBatchlogEndpointsRandom(false, LOCAL, endpoints, Collections::reverse, isAlive, (size) -> size - 1); + Assert.assertEquals(0, result.size()); + } + } + + @Test + public void shouldNotFailIfThereAreAtLeastTwoLiveNodesBesideCoordinator() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 3), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15) + ), this::shouldNotFailIfThereAreAtLeastTwoLiveNodesBesideCoordinator); + } + + private void shouldNotFailIfThereAreAtLeastTwoLiveNodesBesideCoordinator(Multimap endpoints) + { + Predicate isAlive = x -> nodeInRack(x) >= endpoints.get(LOCAL).size() - 2; + for (int i = 0; i < repetitions; i++) + { + if (DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores) { + ReplicaPlans.filterBatchlogEndpointsDynamic(false, LOCAL, endpoints, isAlive); + } else { + ReplicaPlans.filterBatchlogEndpointsRandom(false, LOCAL, endpoints, Collections::reverse, isAlive, (size) -> size - 1); + } + } + } + + @Test + public void shouldNotFailIfThereAreAtLeastTwoLiveNodesIncludingCoordinator() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 3), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 3), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Collections.singletonList(LOCAL), + 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1"), + 15, 15), + + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.prefer_local, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, false, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15) + ), this::shouldNotFailIfThereAreAtLeastTwoLiveNodesIncludingCoordinator); + } + + private void shouldNotFailIfThereAreAtLeastTwoLiveNodesIncludingCoordinator(Multimap endpoints) + { + Predicate isAlive = x -> nodeInRack(x) <= 1; + for (int i = 0; i < repetitions; i++) + { + if (DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores) { + ReplicaPlans.filterBatchlogEndpointsDynamic(false, LOCAL, endpoints, isAlive); + } else { + ReplicaPlans.filterBatchlogEndpointsRandom(false, LOCAL, endpoints, Collections::reverse, isAlive, (size) -> size - 1); + } + } + } + + @Test + public void shouldSelectTwoHostsFromNonLocalRacks() + { + withConfigs(Stream.of( + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Arrays.asList(LOCAL, "1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.random_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "1", "2"), + 15, 1, 1), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "1"), + 15, 15), + () -> configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 15) + ), this::assertTwoEndpointsWithoutCoordinator); + } + + private void assertTwoEndpointsWithoutCoordinator(Multimap endpoints) + { + for (int i = 0; i < repetitions; i++) + { + Collection result = filterBatchlogEndpointsDynamicForTests(endpoints); + // result should be the last two non-local replicas + // (Collections.shuffle has been replaced with Collections.reverse for testing) + assertThat(result.size(), is(2)); + assertThat(new HashSet<>(result).size(), is(2)); + assertThat(result, not(hasItems(endpoints.get(LOCAL).toArray(INET_ADDRESSES)))); + } + } + + /** + * Test with {@link Config.BatchlogEndpointStrategy#dynamic}. + */ + @Test + public void shouldSelectTwoFastestHostsFromSingleLocalRackWithDynamicSnitch() + { + for (int i = 0; i < repetitions; i++) + { + InetAddressAndPort host1 = endpointAddress(0, 1); + InetAddressAndPort host2 = endpointAddress(0, 2); + InetAddressAndPort host3 = endpointAddress(0, 3); + List hosts = Arrays.asList(host1, host2, host3); + + Multimap endpoints = configure(Config.BatchlogEndpointStrategy.dynamic, true, + Collections.singletonList(LOCAL), + 20); + + // ascending + setScores(endpoints, hosts, 10, 12, 14); + List order = Arrays.asList(host1, host2, host3); + assertEquals(order, ReplicaPlans.sortByProximity(Arrays.asList(host3, host1, host2))); + + Collection result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(host1)); + assertThat(result, hasItem(host2)); + + // descending + setScores(endpoints, hosts, 50, 9, 1); + order = Arrays.asList(host3, host2, host1); + assertEquals(order, ReplicaPlans.sortByProximity(Arrays.asList(host1, host2, host3))); + result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(host2)); + assertThat(result, hasItem(host3)); + } + } + + /** + * Test with {@link Config.BatchlogEndpointStrategy#dynamic}. + */ + @Test + public void shouldSelectOneFastestHostsFromNonLocalRackWithDynamicSnitch() + { + for (int i = 0; i < repetitions; i++) + { + // for each rack, get last host (only in test), then sort all endpoints from each rack by scores + Multimap endpoints = configure(Config.BatchlogEndpointStrategy.dynamic, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15); + InetAddressAndPort r0h1 = endpointAddress(0, 1); + InetAddressAndPort r1h1 = endpointAddress(1, 0); + InetAddressAndPort r1h2 = endpointAddress(1, 1); + InetAddressAndPort r2h1 = endpointAddress(2, 0); + InetAddressAndPort r2h2 = endpointAddress(2, 1); + List hosts = Arrays.asList(r0h1, r1h1, r1h2, r2h1, r2h2); + + // ascending + setScores(endpoints, hosts, 11, 6/* r1h1 */, 12, 5/* r2h1 */, 10); + Collection result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(r1h1)); + assertThat(result, hasItem(r2h1)); + + // descending + setScores(endpoints, hosts, 5/* r0h1 */, 20, 5, 0/* r2h1 */, 15); + result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(r0h1)); + assertThat(result, hasItem(r2h1)); + } + } + + /** + * Test with {@link Config.BatchlogEndpointStrategy#dynamic_remote}. + */ + @Test + public void shouldSelectTwoFastestHostsFromSingleLocalRackWithDynamicSnitchRemote() + { + for (int i = 0; i < repetitions; i++) + { + InetAddressAndPort host1 = endpointAddress(0, 1); + InetAddressAndPort host2 = endpointAddress(0, 2); + InetAddressAndPort host3 = endpointAddress(0, 3); + List hosts = Arrays.asList(host1, host2, host3); + + Multimap endpoints = configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Collections.singletonList(LOCAL), + 20); + + // ascending + setScores(endpoints, hosts, + 10, 12, 14); + List order = Arrays.asList(host1, host2, host3); + assertEquals(order, ReplicaPlans.sortByProximity(Arrays.asList(host3, host1, host2))); + + Collection result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(host1)); + assertThat(result, hasItem(host2)); + + // descending + setScores(endpoints, hosts, + 50, 9, 1); + order = Arrays.asList(host3, host2, host1); + assertEquals(order, ReplicaPlans.sortByProximity(Arrays.asList(host1, host2, host3))); + + result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(host2)); + assertThat(result, hasItem(host3)); + } + } + + /** + * Test with {@link Config.BatchlogEndpointStrategy#dynamic_remote}. + */ + @Test + public void shouldSelectOneFastestHostsFromNonLocalRackWithDynamicSnitchRemote() + { + for (int i = 0; i < repetitions; i++) + { + // for each rack, get last host (only in test), then sort all endpoints from each rack by scores + Multimap endpoints = configure(Config.BatchlogEndpointStrategy.dynamic_remote, true, + Arrays.asList(LOCAL, "r1", "r2"), + 15, 15, 15); + InetAddressAndPort r0h1 = endpointAddress(0, 1); + InetAddressAndPort r1h1 = endpointAddress(1, 0); + InetAddressAndPort r1h2 = endpointAddress(1, 1); + InetAddressAndPort r2h1 = endpointAddress(2, 0); + InetAddressAndPort r2h2 = endpointAddress(2, 1); + List hosts = Arrays.asList(r0h1, r1h1, r1h2, r2h1, r2h2); + + // ascending + setScores(endpoints, hosts, + 1, + 10, 12, + 5, 10); + + Collection result = filterBatchlogEndpointsDynamicForTests(endpoints); + filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(r1h1)); + assertThat(result, hasItem(r2h1)); + + // descending + setScores(endpoints, hosts, + 1, // rack 0 + 20, 5, // rack 1 + 0, 15); // rack 2 + result = filterBatchlogEndpointsDynamicForTests(endpoints); + assertThat(result.size(), is(2)); + assertThat(result, hasItem(r1h2)); + assertThat(result, hasItem(r2h1)); + } + } + + private void setScores(Multimap endpoints, + List hosts, + Integer... scores) + { + int maxScore = 0; + + // set the requested scores for the requested hosts + for (int round = 0; round < 50; round++) + { + for (int i = 0; i < hosts.size(); i++) + { + dsnitch.receiveTiming(hosts.get(i), scores[i], MILLISECONDS); + maxScore = Math.max(maxScore, scores[i]); + } + } + + // set some random (higher) scores for unrequested hosts + for (InetAddressAndPort ep : endpoints.values()) + { + if (hosts.contains(ep)) + continue; + for (int r = 0; r < 1; r++) + dsnitch.receiveTiming(ep, maxScore + ThreadLocalRandom.current().nextInt(100) + 1, MILLISECONDS); + } + + dsnitch.updateScores(); + } + + private int nodeInRack(InetAddressAndPort input) + { + return input.address.getAddress()[3]; + } + + private static InetAddressAndPort endpointAddress(int rack, int nodeInRack) + { + if (rack == 0 && nodeInRack == 0) + return FBUtilities.getBroadcastAddressAndPort(); + + try + { + return InetAddressAndPort.getByAddress(new byte[]{ 0, 0, (byte) rack, (byte) nodeInRack }); + } + catch (UnknownHostException e) + { + throw new RuntimeException(e); + } + } + + private Multimap configure(Config.BatchlogEndpointStrategy batchlogEndpointStrategy, + boolean dynamicSnitch, + List racks, + int... nodesPerRack) + { + DatabaseDescriptor.setDynamicBadnessThreshold(0.1); + StorageService.instance.unsafeInitialize(); + + // if any of the three assertions fires, your test is busted + assert !racks.isEmpty(); + assert racks.size() <= 10; + assert racks.size() == nodesPerRack.length; + + ImmutableMultimap.Builder builder = ImmutableMultimap.builder(); + for (int r = 0; r < racks.size(); r++) + { + String rack = racks.get(r); + for (int n = 0; n < nodesPerRack[r]; n++) + builder.put(rack, endpointAddress(r, n)); + } + + ImmutableMultimap endpoints = builder.build(); + + reconfigure(batchlogEndpointStrategy, dynamicSnitch, endpoints); + + return endpoints; + } + + private void reconfigure(Config.BatchlogEndpointStrategy batchlogEndpointStrategy, + boolean dynamicSnitch, + Multimap endpoints) + { + DatabaseDescriptor.setBatchlogEndpointStrategy(batchlogEndpointStrategy); + + if (DatabaseDescriptor.getEndpointSnitch() instanceof DynamicEndpointSnitch) + ((DynamicEndpointSnitch) DatabaseDescriptor.getEndpointSnitch()).close(); + + Multimap endpointRacks = Multimaps.invertFrom(endpoints, ArrayListMultimap.create()); + GossipingPropertyFileSnitch gpfs = new GossipingPropertyFileSnitch() + { + @Override + public String getDatacenter(InetAddressAndPort endpoint) + { + return "dc1"; + } + + @Override + public String getRack(InetAddressAndPort endpoint) + { + return endpointRacks.get(endpoint).iterator().next(); + } + }; + IEndpointSnitch snitch; + if (dynamicSnitch) + snitch = dsnitch = new DynamicEndpointSnitch(gpfs, String.valueOf(gpfs.hashCode())); + else + { + dsnitch = null; + snitch = gpfs; + } + + DatabaseDescriptor.setDynamicBadnessThreshold(0); + DatabaseDescriptor.setEndpointSnitch(snitch); + + DatabaseDescriptor.setBatchlogEndpointStrategy(batchlogEndpointStrategy); + } + + private void withConfigs(Stream>> supplierStream, + Consumer> testFunction) + { + supplierStream.map(Supplier::get) + .forEach(endpoints -> { + try + { + testFunction.accept(endpoints); + } + catch (AssertionError e) + { + throw new AssertionError(configToString(endpoints), e); + } + }); + } + + private String configToString(Multimap endpoints) + { + return "strategy:" + DatabaseDescriptor.getBatchlogEndpointStrategy() + + " snitch:" + DatabaseDescriptor.getEndpointSnitch().getClass().getSimpleName() + + " nodes-per-rack: " + endpoints.asMap().entrySet().stream() + .map(e -> e.getKey() + '=' + e.getValue().size()) + .collect(Collectors.joining()); + } + } diff --git a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java index a832679422..89c6e7ad89 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -80,6 +80,7 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.config.Config$RepairCommandPoolFullStrategy", "org.apache.cassandra.config.Config$UserFunctionTimeoutPolicy", "org.apache.cassandra.config.Config$CorruptedTombstoneStrategy", + "org.apache.cassandra.config.Config$BatchlogEndpointStrategy", "org.apache.cassandra.config.DatabaseDescriptor$ByteUnit", "org.apache.cassandra.config.ParameterizedClass", "org.apache.cassandra.config.EncryptionOptions",