diff --git a/CHANGES.txt b/CHANGES.txt index 6d0267b843..23bb86b090 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -105,6 +105,7 @@ Merged from 5.0: Merged from 4.1: * Fix race condition in DecayingEstimatedHistogramReservoir during rescale (CASSANDRA-19365) Merged from 4.0: + * 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 1c53401057..ef450de7fd 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -156,6 +156,32 @@ auto_hints_cleanup_enabled: false # Min unit: KiB batchlog_replay_throttle: 1024KiB +# 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 +# Recommended, 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: dynamic_remote + # Authentication backend, implementing IAuthenticator; used to identify users # Out of the box, Cassandra provides org.apache.cassandra.auth.{AllowAllAuthenticator, # PasswordAuthenticator}. diff --git a/conf/cassandra_latest.yaml b/conf/cassandra_latest.yaml index 71ceaadaa3..c9eb784a23 100644 --- a/conf/cassandra_latest.yaml +++ b/conf/cassandra_latest.yaml @@ -159,6 +159,32 @@ auto_hints_cleanup_enabled: false # Min unit: KiB batchlog_replay_throttle: 1024KiB +# 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: dynamic_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 56a8aad315..b98d1b2d9d 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -418,6 +418,10 @@ public enum CassandraRelevantProperties REPLACE_ADDRESS_FIRST_BOOT("cassandra.replace_address_first_boot"), REPLACE_NODE("cassandra.replace_node"), REPLACE_TOKEN("cassandra.replace_token"), + /** + * 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"), /** * Whether we reset any found data from previously run bootstraps. */ diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index b9daf8af05..a5d18386e1 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -440,6 +440,7 @@ public class Config public DataStorageSpec.IntKibibytesBound hinted_handoff_throttle = new DataStorageSpec.IntKibibytesBound("1024KiB"); @Replaces(oldName = "batchlog_replay_throttle_in_kb", converter = Converters.KIBIBYTES_DATASTORAGE, deprecated = true) public DataStorageSpec.IntKibibytesBound batchlog_replay_throttle = new DataStorageSpec.IntKibibytesBound("1024KiB"); + public BatchlogEndpointStrategy batchlog_endpoint_strategy = BatchlogEndpointStrategy.random_remote; public int max_hints_delivery_threads = 2; @Replaces(oldName = "hints_flush_period_in_ms", converter = Converters.MILLIS_DURATION_INT, deprecated = true) public DurationSpec.IntMillisecondsBound hints_flush_period = new DurationSpec.IntMillisecondsBound("10s"); @@ -1253,6 +1254,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 Set SENSITIVE_KEYS = new HashSet() {{ 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 7009da0aba..4024d0ea0f 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -3677,6 +3677,22 @@ public class DatabaseDescriptor conf.batchlog_replay_throttle = new DataStorageSpec.IntKibibytesBound(throttleInKiB); } + 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 77e04e6887..bb652b6cff 100644 --- a/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java +++ b/src/java/org/apache/cassandra/locator/DynamicEndpointSnitch.java @@ -29,8 +29,8 @@ import java.util.stream.Collectors; import com.google.common.annotations.VisibleForTesting; 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; diff --git a/src/java/org/apache/cassandra/locator/ReplicaPlans.java b/src/java/org/apache/cassandra/locator/ReplicaPlans.java index f17909ae98..51830ffa4b 100644 --- a/src/java/org/apache/cassandra/locator/ReplicaPlans.java +++ b/src/java/org/apache/cassandra/locator/ReplicaPlans.java @@ -18,16 +18,37 @@ 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 javax.annotation.Nullable; + 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; @@ -49,19 +70,6 @@ import org.apache.cassandra.service.reads.SpeculativeRetryPolicy; import org.apache.cassandra.tcm.Epoch; import org.apache.cassandra.tcm.membership.NodeState; 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.Set; -import java.util.concurrent.ThreadLocalRandom; -import java.util.function.*; - -import javax.annotation.Nullable; import static com.google.common.collect.Iterables.any; import static com.google.common.collect.Iterables.filter; @@ -80,6 +88,16 @@ public class ReplicaPlans private static final Range FULL_TOKEN_RANGE = new Range<>(DatabaseDescriptor.getPartitioner().getMinimumToken(), DatabaseDescriptor.getPartitioner().getMinimumToken()); + 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) @@ -286,14 +304,27 @@ 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(), + Collection chosenEndpoints = filterBatchlogEndpoints(false, + snitch.getLocalRack(), localEndpoints, Collections::shuffle, (r) -> FailureDetector.isEndpointAlive.test(r) && metadata.directory.peerState(r) == NodeState.JOINED, ThreadLocalRandom.current()::nextInt); - 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); + } return ReplicaLayout.forTokenWrite(Keyspace.open(SchemaConstants.SYSTEM_KEYSPACE_NAME).getReplicationStrategy(), SystemReplicas.getSystemReplicas(chosenEndpoints).forToken(token), @@ -335,18 +366,30 @@ public class ReplicaPlans // 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, + public static Collection filterBatchlogEndpoints(boolean preferLocalRack, + String localRack, Multimap endpoints, Consumer> shuffle, Predicate include, Function indexPicker) { + return DatabaseDescriptor.getBatchlogEndpointStrategy().useDynamicSnitchScores && DatabaseDescriptor.isDynamicEndpointSnitch() + ? filterBatchlogEndpointsDynamic(preferLocalRack, localRack, endpoints, FailureDetector.isEndpointAlive) + : filterBatchlogEndpointsRandom(preferLocalRack, localRack, endpoints, Collections::shuffle, FailureDetector.isEndpointAlive, ThreadLocalRandom.current()::nextInt); + } + + private static ListMultimap validate(boolean preferLocalRack, String localRack, + Multimap endpoints, + Predicate include) + { + 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(); @@ -354,15 +397,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 include, + Function indexPicker) + { + ListMultimap validated = validate(preferLocalRack, localRack, endpoints, include); + + // 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) { /* @@ -372,24 +445,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()))); @@ -398,6 +479,56 @@ public class ReplicaPlans return result; } + @VisibleForTesting + public static Collection filterBatchlogEndpointsDynamic(boolean preferLocalRack, String localRack, + Multimap endpoints, + Predicate include) + { + ListMultimap validated = validate(preferLocalRack, localRack, endpoints, include); + + // 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.ForWrite forReadRepair(ReplicaPlan forRead, ClusterMetadata metadata, Keyspace keyspace, ConsistencyLevel consistencyLevel, Token token, Predicate isAlive) throws UnavailableException { AbstractReplicationStrategy replicationStrategy = keyspace.getReplicationStrategy(); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index a78786fb3f..3b502f5a48 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -4424,6 +4424,16 @@ public class StorageService extends NotificationBroadcasterSupport implements IE } } + public String getBatchlogEndpointStrategy() + { + return DatabaseDescriptor.getBatchlogEndpointStrategy().name(); + } + + public void setBatchlogEndpointStrategy(String batchlogEndpointStrategy) + { + DatabaseDescriptor.setBatchlogEndpointStrategy(Config.BatchlogEndpointStrategy.valueOf(batchlogEndpointStrategy)); + } + public StreamStateStore streamStateStore() { return streamStateStore; diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index da4206416f..8beeb32ba5 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -653,6 +653,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 6d9a52fb65..c5041e6e70 100644 --- a/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java +++ b/test/unit/org/apache/cassandra/batchlog/BatchlogEndpointFilterTest.java @@ -18,25 +18,54 @@ 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.After; +import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; +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.tcm.ClusterMetadataService; +import org.apache.cassandra.tcm.StubClusterMetadataService; +import org.apache.cassandra.utils.FBUtilities; -import static org.hamcrest.CoreMatchers.is; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.assertj.core.api.Assertions.assertThat; +import static org.hamcrest.CoreMatchers.*; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; + public class BatchlogEndpointFilterTest { private static final String LOCAL = "local"; + private double oldBadness; @BeforeClass public static void initialiseServer() @@ -44,9 +73,51 @@ public class BatchlogEndpointFilterTest DatabaseDescriptor.daemonInitialization(); } - @Test - public void shouldSelect2HostsFromNonLocalRacks() throws UnknownHostException + @Before + public void before() { + oldBadness = DatabaseDescriptor.getDynamicBadnessThreshold(); + DatabaseDescriptor.setDynamicBadnessThreshold(0.1); + ClusterMetadataService.unsetInstance(); + ClusterMetadataService.setInstance(StubClusterMetadataService.forTesting()); + StorageService.instance.unsafeInitialize(); + } + + @After + public void after() + { + DatabaseDescriptor.setDynamicBadnessThreshold(oldBadness); + } + + // 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; + + @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")) @@ -55,8 +126,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"))); } @@ -64,6 +154,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")) @@ -73,8 +164,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) @@ -85,13 +177,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"))); } @@ -99,17 +193,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")) @@ -117,7 +214,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)); @@ -128,13 +226,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)); @@ -145,24 +245,811 @@ 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.addressBytes[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) + { + // 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 840ef49d71..27d4d19528 100644 --- a/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java +++ b/test/unit/org/apache/cassandra/config/DatabaseDescriptorRefTest.java @@ -101,6 +101,9 @@ public class DatabaseDescriptorRefTest "org.apache.cassandra.config.ConfigBeanInfo", "org.apache.cassandra.config.ConfigCustomizer", "org.apache.cassandra.config.ConfigurationLoader", + "org.apache.cassandra.config.Config$CorruptedTombstoneStrategy", + "org.apache.cassandra.config.Config$BatchlogEndpointStrategy", + "org.apache.cassandra.config.DatabaseDescriptor$ByteUnit", "org.apache.cassandra.config.DataRateSpec", "org.apache.cassandra.config.DataRateSpec$DataRateUnit", "org.apache.cassandra.config.DataRateSpec$DataRateUnit$1",