From 2346ed8241022882e77433e283ab8ce3004d12b0 Mon Sep 17 00:00:00 2001 From: Mick Semb Wever Date: Wed, 2 Dec 2020 09:34:28 +0100 Subject: [PATCH] Offline token allocation strategy generator tool Provides the tools/bin/generate-tokens script that can be used to pregenerate allocation strategy tokens. Refactors TokenAllocation for extensibility (and better re-use between strategies), and adds OfflineTokenAllocator. TokenMetadata now has a configurable snitch (instead of hardcoded to DatabaseDescriptor.getEndpointSnitch()) for testability. Adds logging of growth and stddev changes to allocate_tokens_for_local_replication_factor usages. patch by Mick Semb Wever; reviewed by Paul Motta, Ekaterina Dimitrova for CASSANDRA-16205 --- CHANGES.txt | 1 + debian/cassandra-tools.install | 3 +- doc/source/tools/generatetokens.rst | 56 +++ doc/source/tools/index.rst | 1 + redhat/cassandra.spec | 2 + .../tokenallocator/OfflineTokenAllocator.java | 190 ++++++++ .../dht/tokenallocator/TokenAllocation.java | 427 ++++++++---------- .../locator/AbstractReplicationStrategy.java | 10 - .../cassandra/locator/SimpleSnitch.java | 4 +- .../cassandra/locator/TokenMetadata.java | 33 +- .../cassandra/tools/GenerateTokens.java | 157 +++++++ .../apache/cassandra/utils/OutputHandler.java | 13 +- .../cassandra/dht/BootStrapperTest.java | 213 +-------- .../OfflineTokenAllocatorTest.java | 222 +++++++++ .../tokenallocator/TokenAllocationTest.java | 310 +++++++++++++ .../cassandra/tools/GenerateTokensTest.java | 56 +++ tools/bin/generatetokens | 54 +++ 17 files changed, 1292 insertions(+), 460 deletions(-) create mode 100644 doc/source/tools/generatetokens.rst create mode 100644 src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java create mode 100644 src/java/org/apache/cassandra/tools/GenerateTokens.java create mode 100644 test/unit/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocatorTest.java create mode 100644 test/unit/org/apache/cassandra/dht/tokenallocator/TokenAllocationTest.java create mode 100644 test/unit/org/apache/cassandra/tools/GenerateTokensTest.java create mode 100755 tools/bin/generatetokens diff --git a/CHANGES.txt b/CHANGES.txt index 172ebcc2e1..52d2526168 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0-beta4 + * Add generatetokens script for offline token allocation strategy generation (CASSANDRA-16205) * Remove Windows scripts (CASSANDRA-16171) * Improve checksumming and compression in protocol V5 (CASSANDRA-15299) * Optimised repair streaming improvements (CASSANDRA-16274) diff --git a/debian/cassandra-tools.install b/debian/cassandra-tools.install index 88063441ea..7cebbefa5a 100644 --- a/debian/cassandra-tools.install +++ b/debian/cassandra-tools.install @@ -1,7 +1,8 @@ +tools/bin/generatetokens usr/bin +tools/bin/sstabledump usr/bin tools/bin/sstableexpiredblockers usr/bin tools/bin/sstablelevelreset usr/bin tools/bin/sstablemetadata usr/bin tools/bin/sstableofflinerelevel usr/bin tools/bin/sstablerepairedset usr/bin tools/bin/sstablesplit usr/bin -tools/bin/sstabledump usr/bin diff --git a/doc/source/tools/generatetokens.rst b/doc/source/tools/generatetokens.rst new file mode 100644 index 0000000000..24448d0438 --- /dev/null +++ b/doc/source/tools/generatetokens.rst @@ -0,0 +1,56 @@ +.. Licensed to the Apache Software Foundation (ASF) under one +.. or more contributor license agreements. See the NOTICE file +.. distributed with this work for additional information +.. regarding copyright ownership. The ASF licenses this file +.. to you under the Apache License, Version 2.0 (the +.. "License"); you may not use this file except in compliance +.. with the License. You may obtain a copy of the License at +.. +.. http://www.apache.org/licenses/LICENSE-2.0 +.. +.. Unless required by applicable law or agreed to in writing, software +.. distributed under the License is distributed on an "AS IS" BASIS, +.. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +.. See the License for the specific language governing permissions and +.. limitations under the License. + +generatetokens +------------ + +Pre-generates tokens for a datacenter with the given number of nodes using the token allocation algorithm. Useful in edge-cases when generated tokens needs to be known in advance of bootstrapping nodes. In nearly all cases it is best to just let the bootstrapping nodes automatically generate their own tokens. +ref: https://issues.apache.org/jira/browse/CASSANDRA-16205 + + +Usage +^^^^^ +generatetokens -n NODES -t TOKENS --rf REPLICATION_FACTOR [--partitioner PARTITIONER] [--racks RACK_NODE_COUNTS] + + +=================================== ================================================================================ + -n,--nodes Number of nodes. + -t,--tokens Number of tokens/vnodes per node. + --rf Replication factor. + -p,--partitioner Database partitioner, either Murmur3Partitioner or RandomPartitioner. + --racks Number of nodes per rack, separated by commas. Must add up to the total node count. For example, 'generatetokens -n 30 -t 8 --rf 3 --racks 10,10,10' will generate tokens for three racks of 10 nodes each. +=================================== ================================================================================ + + +This command, if used, is expected to be run before the Cassandra node is first started. The output from the command is used to configure the nodes `num_tokens` setting in the `cassandra.yaml` + + +Example Output +^^^^^^^^^^^^^^ +Example usage and output is + + $ tools/bin/generatetokens -n 9 -t 4 --rf 3 --racks 3,3,3 + + Generating tokens for 9 nodes with 4 vnodes each for replication factor 3 and partitioner Murmur3Partitioner + Node 0 rack 0: [-6270077235120413733, -1459727275878514299, 2887564907718879562, 5778609289102954400] + Node 1 rack 1: [-8780789453057732897, -3279530982831298765, 1242905755717369197, 8125606440590916903] + Node 2 rack 2: [-7240625170832344686, -4453155190605073029, 74749827226930055, 4615117940688406403] + Node 3 rack 0: [-5361616212862743381, -2369629129354906532, 2065235331718124379, 6952107864846935651] + Node 4 rack 1: [-8010707311945038792, -692488724325792122, 3751341424203642982, 7538857152718926277] + Node 5 rack 2: [-7625666241388691739, -3866343086718185897, 5196863614895680401, 8895780530621367810] + Node 6 rack 0: [-5815846723991578557, -1076108000102153211, 1654070543717746788, 8510693485606142356] + Node 7 rack 1: [-2824580056093102649, 658827791472149626, 3319453165961261272, 6365358576974945025] + Node 8 rack 2: [-4159749138661629463, -1914678202616710416, 4905990777792043402, 6658733220910940338] \ No newline at end of file diff --git a/doc/source/tools/index.rst b/doc/source/tools/index.rst index d28929c842..56fbabc0e0 100644 --- a/doc/source/tools/index.rst +++ b/doc/source/tools/index.rst @@ -23,6 +23,7 @@ This section describes the command line tools provided with Apache Cassandra. :maxdepth: 3 cqlsh + generatetokens nodetool/nodetool sstable/index cassandra_stress diff --git a/redhat/cassandra.spec b/redhat/cassandra.spec index f0652c4bde..dcbda87776 100644 --- a/redhat/cassandra.spec +++ b/redhat/cassandra.spec @@ -126,6 +126,7 @@ exit 0 %attr(755,root,root) %{_bindir}/cqlsh.py %attr(755,root,root) %{_bindir}/debug-cql %attr(755,root,root) %{_bindir}/fqltool +%attr(755,root,root) %{_bindir}/generatetokens %attr(755,root,root) %{_bindir}/nodetool %attr(755,root,root) %{_bindir}/sstableloader %attr(755,root,root) %{_bindir}/sstablescrub @@ -180,6 +181,7 @@ This package contains extra tools for working with Cassandra clusters. %attr(755,root,root) %{_bindir}/auditlogviewer %attr(755,root,root) %{_bindir}/jmxtool %attr(755,root,root) %{_bindir}/fqltool +%attr(755,root,root) %{_bindir}/generatetokens %changelog diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java b/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java new file mode 100644 index 0000000000..f8c13df513 --- /dev/null +++ b/src/java/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocator.java @@ -0,0 +1,190 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.dht.tokenallocator; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.TreeSet; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import org.apache.commons.math3.stat.descriptive.SummaryStatistics; + +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.SimpleSnitch; +import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.utils.OutputHandler; + +public class OfflineTokenAllocator +{ + public static List allocate(int rf, int numTokens, int[] nodesPerRack, OutputHandler logger, IPartitioner partitioner) + { + Preconditions.checkArgument(rf > 0, "rf must be greater than zero"); + Preconditions.checkArgument(numTokens > 0, "num_tokens must be greater than zero"); + Preconditions.checkNotNull(nodesPerRack); + Preconditions.checkArgument(nodesPerRack.length > 0, "nodesPerRack must contain a node count for at least one rack"); + Preconditions.checkNotNull(logger); + Preconditions.checkNotNull(partitioner); + + int nodes = Arrays.stream(nodesPerRack).sum(); + + Preconditions.checkArgument(nodes >= rf, + "not enough nodes %s for rf %s in %s", Arrays.stream(nodesPerRack).sum(), rf, Arrays.toString(nodesPerRack)); + + List fakeNodes = new ArrayList<>(nodes); + MultinodeAllocator allocator = new MultinodeAllocator(rf, numTokens, logger, partitioner); + + // Defensive-copy method argument + nodesPerRack = Arrays.copyOf(nodesPerRack, nodesPerRack.length); + + int racks = nodesPerRack.length; + int nodeId = 0; + int rackId = 0; + // Distribute nodes among the racks in round-robin fashion in the order the user is supposed to start them. + while (nodesPerRack[rackId] > 0) + { + // Allocate tokens for current node + fakeNodes.add(allocator.allocateTokensForNode(nodeId++, rackId)); + + // Find next rack with unallocated node + int nextRack = (rackId+1) % racks; + while (nodesPerRack[nextRack] == 0 && nextRack != rackId) + nextRack = (nextRack+1) % racks; + + // Update nodesPerRack and rackId + nodesPerRack[rackId]--; + rackId = nextRack; + } + return fakeNodes; + } + + public static class FakeNode + { + private final InetAddressAndPort fakeAddressAndPort; + private final int rackId; + private final Collection tokens; + + public FakeNode(InetAddressAndPort address, Integer rackId, Collection tokens) + { + this.fakeAddressAndPort = address; + this.rackId = rackId; + // Sort tokens for better presentation + this.tokens = new TreeSet<>(tokens); + } + + public int nodeId() + { + return fakeAddressAndPort.port; + } + + public int rackId() + { + return rackId; + } + + public Collection tokens() + { + return tokens; + } + } + + private static class MultinodeAllocator + { + private final FakeSnitch fakeSnitch; + private final TokenMetadata fakeMetadata; + private final TokenAllocation allocation; + private final Map lastCheckPoint = Maps.newHashMap(); + private final OutputHandler logger; + + private MultinodeAllocator(int rf, int numTokens, OutputHandler logger, IPartitioner partitioner) + { + this.fakeSnitch = new FakeSnitch(); + this.fakeMetadata = new TokenMetadata(fakeSnitch).cloneWithNewPartitioner(partitioner); + this.allocation = TokenAllocation.create(fakeSnitch, fakeMetadata, rf, numTokens); + this.logger = logger; + } + + private FakeNode allocateTokensForNode(int nodeId, Integer rackId) + { + // Update snitch and token metadata info + InetAddressAndPort fakeNodeAddressAndPort = getLoopbackAddressWithPort(nodeId); + fakeSnitch.nodeByRack.put(fakeNodeAddressAndPort, rackId); + fakeMetadata.updateTopology(fakeNodeAddressAndPort); + + // Allocate tokens + Collection tokens = allocation.allocate(fakeNodeAddressAndPort); + + // Validate ownership stats + validateAllocation(nodeId, rackId); + + return new FakeNode(fakeNodeAddressAndPort, rackId, tokens); + } + + private void validateAllocation(int nodeId, int rackId) + { + SummaryStatistics newOwnership = allocation.getAllocationRingOwnership(SimpleSnitch.DATA_CENTER_NAME, Integer.toString(rackId)); + SummaryStatistics oldOwnership = lastCheckPoint.put(rackId, newOwnership); + if (oldOwnership != null) + logger.debug(String.format("Replicated node load in rack=%d before allocating node %d: %s.", rackId, nodeId, + TokenAllocation.statToString(oldOwnership))); + logger.debug(String.format("Replicated node load in rack=%d after allocating node %d: %s.", rackId, nodeId, + TokenAllocation.statToString(newOwnership))); + if (oldOwnership != null && oldOwnership.getStandardDeviation() != 0.0) + { + double stdDevGrowth = newOwnership.getStandardDeviation() - oldOwnership.getStandardDeviation(); + if (stdDevGrowth > TokenAllocation.WARN_STDEV_GROWTH) + { + logger.warn(String.format("Growth of %.2f%% in token ownership standard deviation after allocating node %d on rack %d above warning threshold of %d%%", + stdDevGrowth * 100, nodeId, rackId, (int)(TokenAllocation.WARN_STDEV_GROWTH * 100))); + } + } + } + } + + private static class FakeSnitch extends SimpleSnitch + { + final Map nodeByRack = new HashMap<>(); + + @Override + public String getRack(InetAddressAndPort endpoint) + { + return Integer.toString(nodeByRack.get(endpoint)); + } + } + + private static InetAddressAndPort getLoopbackAddressWithPort(int port) + { + try + { + return InetAddressAndPort.getByAddressOverrideDefaults(InetAddress.getByName("127.0.0.1"), port); + } + catch (UnknownHostException e) + { + throw new IllegalStateException("Unexpected UnknownHostException", e); + } + } +} \ No newline at end of file diff --git a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java index bd6d9802db..e3e81dc116 100644 --- a/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java +++ b/src/java/org/apache/cassandra/dht/tokenallocator/TokenAllocation.java @@ -18,6 +18,7 @@ package org.apache.cassandra.dht.tokenallocator; import java.util.Collection; +import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; @@ -45,32 +46,27 @@ import org.apache.cassandra.locator.TokenMetadata.Topology; public class TokenAllocation { + public static final double WARN_STDEV_GROWTH = 0.05; + private static final Logger logger = LoggerFactory.getLogger(TokenAllocation.class); + final TokenMetadata tokenMetadata; + final AbstractReplicationStrategy replicationStrategy; + final int numTokens; + final Map> strategyByRackDc = new HashMap<>(); + + private TokenAllocation(TokenMetadata tokenMetadata, AbstractReplicationStrategy replicationStrategy, int numTokens) + { + this.tokenMetadata = tokenMetadata.cloneOnlyTokenMap(); + this.replicationStrategy = replicationStrategy; + this.numTokens = numTokens; + } public static Collection allocateTokens(final TokenMetadata tokenMetadata, final AbstractReplicationStrategy rs, final InetAddressAndPort endpoint, int numTokens) { - TokenMetadata tokenMetadataCopy = tokenMetadata.cloneOnlyTokenMap(); - StrategyAdapter strategy = getStrategy(tokenMetadataCopy, rs, endpoint); - Collection tokens = create(tokenMetadata, strategy).addUnit(endpoint, numTokens); - tokens = adjustForCrossDatacenterClashes(tokenMetadata, strategy, tokens); - - if (logger.isWarnEnabled()) - { - logger.warn("Selected tokens {}", tokens); - SummaryStatistics os = replicatedOwnershipStats(tokenMetadataCopy, rs, endpoint); - tokenMetadataCopy.updateNormalTokens(tokens, endpoint); - SummaryStatistics ns = replicatedOwnershipStats(tokenMetadataCopy, rs, endpoint); - logger.warn("Replicated node load in datacenter before allocation {}", statToString(os)); - logger.warn("Replicated node load in datacenter after allocation {}", statToString(ns)); - - // TODO: Is it worth doing the replicated ownership calculation always to be able to raise this alarm? - if (ns.getStandardDeviation() > os.getStandardDeviation()) - logger.warn("Unexpected growth in standard deviation after allocation."); - } - return tokens; + return create(tokenMetadata, rs, numTokens).allocate(endpoint); } public static Collection allocateTokens(final TokenMetadata tokenMetadata, @@ -78,114 +74,209 @@ public class TokenAllocation final InetAddressAndPort endpoint, int numTokens) { - TokenMetadata tokenMetadataCopy = tokenMetadata.cloneOnlyTokenMap(); - StrategyAdapter strategy = getStrategy(tokenMetadataCopy, replicas, endpoint); - Collection tokens = create(tokenMetadata, strategy).addUnit(endpoint, numTokens); - tokens = adjustForCrossDatacenterClashes(tokenMetadata, strategy, tokens); - logger.warn("Selected tokens {}", tokens); - // SummaryStatistics is not implemented for `allocate_tokens_for_local_replication_factor` + return create(DatabaseDescriptor.getEndpointSnitch(), tokenMetadata, replicas, numTokens).allocate(endpoint); + } + + static TokenAllocation create(IEndpointSnitch snitch, TokenMetadata tokenMetadata, int replicas, int numTokens) + { + // We create a fake NTS replication strategy with the specified RF in the local DC + HashMap options = new HashMap<>(); + options.put(snitch.getLocalDatacenter(), Integer.toString(replicas)); + NetworkTopologyStrategy fakeReplicationStrategy = new NetworkTopologyStrategy(null, tokenMetadata, snitch, options); + + TokenAllocation allocator = new TokenAllocation(tokenMetadata, fakeReplicationStrategy, numTokens); + return allocator; + } + + static TokenAllocation create(TokenMetadata tokenMetadata, AbstractReplicationStrategy rs, int numTokens) + { + return new TokenAllocation(tokenMetadata, rs, numTokens); + } + + Collection allocate(InetAddressAndPort endpoint) + { + StrategyAdapter strategy = getOrCreateStrategy(endpoint); + Collection tokens = strategy.createAllocator().addUnit(endpoint, numTokens); + tokens = strategy.adjustForCrossDatacenterClashes(tokens); + + SummaryStatistics os = strategy.replicatedOwnershipStats(); + tokenMetadata.updateNormalTokens(tokens, endpoint); + + SummaryStatistics ns = strategy.replicatedOwnershipStats(); + logger.info("Selected tokens {}", tokens); + logger.debug("Replicated node load in datacenter before allocation {}", statToString(os)); + logger.debug("Replicated node load in datacenter after allocation {}", statToString(ns)); + + double stdDevGrowth = ns.getStandardDeviation() - os.getStandardDeviation(); + if (stdDevGrowth > TokenAllocation.WARN_STDEV_GROWTH) + { + logger.warn(String.format("Growth of %.2f%% in token ownership standard deviation after allocation above warning threshold of %d%%", + stdDevGrowth * 100, (int)(TokenAllocation.WARN_STDEV_GROWTH * 100))); + } + return tokens; } - private static Collection adjustForCrossDatacenterClashes(final TokenMetadata tokenMetadata, - StrategyAdapter strategy, Collection tokens) - { - List filtered = Lists.newArrayListWithCapacity(tokens.size()); - - for (Token t : tokens) - { - while (tokenMetadata.getEndpoint(t) != null) - { - InetAddressAndPort other = tokenMetadata.getEndpoint(t); - if (strategy.inAllocationRing(other)) - throw new ConfigurationException(String.format("Allocated token %s already assigned to node %s. Is another node also allocating tokens?", t, other)); - t = t.increaseSlightly(); - } - filtered.add(t); - } - return filtered; - } - - // return the ratio of ownership for each endpoint - public static Map evaluateReplicatedOwnership(TokenMetadata tokenMetadata, AbstractReplicationStrategy rs) - { - Map ownership = Maps.newHashMap(); - List sortedTokens = tokenMetadata.sortedTokens(); - Iterator it = sortedTokens.iterator(); - Token current = it.next(); - while (it.hasNext()) - { - Token next = it.next(); - addOwnership(tokenMetadata, rs, current, next, ownership); - current = next; - } - addOwnership(tokenMetadata, rs, current, sortedTokens.get(0), ownership); - - return ownership; - } - - static void addOwnership(final TokenMetadata tokenMetadata, final AbstractReplicationStrategy rs, Token current, Token next, Map ownership) - { - double size = current.size(next); - Token representative = current.getPartitioner().midpoint(current, next); - for (InetAddressAndPort n : rs.calculateNaturalReplicas(representative, tokenMetadata).endpoints()) - { - Double v = ownership.get(n); - ownership.put(n, v != null ? v + size : size); - } - } - - public static String statToString(SummaryStatistics stat) + static String statToString(SummaryStatistics stat) { return String.format("max %.2f min %.2f stddev %.4f", stat.getMax() / stat.getMean(), stat.getMin() / stat.getMean(), stat.getStandardDeviation()); } - public static SummaryStatistics replicatedOwnershipStats(TokenMetadata tokenMetadata, - AbstractReplicationStrategy rs, InetAddressAndPort endpoint) + SummaryStatistics getAllocationRingOwnership(String datacenter, String rack) { - SummaryStatistics stat = new SummaryStatistics(); - StrategyAdapter strategy = getStrategy(tokenMetadata, rs, endpoint); - for (Map.Entry en : evaluateReplicatedOwnership(tokenMetadata, rs).entrySet()) - { - // Filter only in the same datacentre. - if (strategy.inAllocationRing(en.getKey())) - stat.addValue(en.getValue() / tokenMetadata.getTokens(en.getKey()).size()); - } - return stat; + return getOrCreateStrategy(datacenter, rack).replicatedOwnershipStats(); } - static TokenAllocator create(TokenMetadata tokenMetadata, StrategyAdapter strategy) + SummaryStatistics getAllocationRingOwnership(InetAddressAndPort endpoint) { - NavigableMap sortedTokens = new TreeMap<>(); - for (Map.Entry en : tokenMetadata.getNormalAndBootstrappingTokenToEndpointMap().entrySet()) - { - if (strategy.inAllocationRing(en.getValue())) - sortedTokens.put(en.getKey(), en.getValue()); - } - return TokenAllocatorFactory.createTokenAllocator(sortedTokens, strategy, tokenMetadata.partitioner); + return getOrCreateStrategy(endpoint).replicatedOwnershipStats(); } - interface StrategyAdapter extends ReplicationStrategy + abstract class StrategyAdapter implements ReplicationStrategy { // return true iff the provided endpoint occurs in the same virtual token-ring we are allocating for // i.e. the set of the nodes that share ownership with the node we are allocating // alternatively: return false if the endpoint's ownership is independent of the node we are allocating tokens for - boolean inAllocationRing(InetAddressAndPort other); + abstract boolean inAllocationRing(InetAddressAndPort other); + + final TokenAllocator createAllocator() + { + NavigableMap sortedTokens = new TreeMap<>(); + for (Map.Entry en : tokenMetadata.getNormalAndBootstrappingTokenToEndpointMap().entrySet()) + { + if (inAllocationRing(en.getValue())) + sortedTokens.put(en.getKey(), en.getValue()); + } + return TokenAllocatorFactory.createTokenAllocator(sortedTokens, this, tokenMetadata.partitioner); + } + + final Collection adjustForCrossDatacenterClashes(Collection tokens) + { + List filtered = Lists.newArrayListWithCapacity(tokens.size()); + + for (Token t : tokens) + { + while (tokenMetadata.getEndpoint(t) != null) + { + InetAddressAndPort other = tokenMetadata.getEndpoint(t); + if (inAllocationRing(other)) + throw new ConfigurationException(String.format("Allocated token %s already assigned to node %s. Is another node also allocating tokens?", t, other)); + t = t.increaseSlightly(); + } + filtered.add(t); + } + return filtered; + } + + final SummaryStatistics replicatedOwnershipStats() + { + SummaryStatistics stat = new SummaryStatistics(); + for (Map.Entry en : evaluateReplicatedOwnership().entrySet()) + { + // Filter only in the same allocation ring + if (inAllocationRing(en.getKey())) + stat.addValue(en.getValue() / tokenMetadata.getTokens(en.getKey()).size()); + } + return stat; + } + + // return the ratio of ownership for each endpoint + private Map evaluateReplicatedOwnership() + { + Map ownership = Maps.newHashMap(); + List sortedTokens = tokenMetadata.sortedTokens(); + if (sortedTokens.isEmpty()) + return ownership; + + Iterator it = sortedTokens.iterator(); + Token current = it.next(); + while (it.hasNext()) + { + Token next = it.next(); + addOwnership(current, next, ownership); + current = next; + } + addOwnership(current, sortedTokens.get(0), ownership); + + return ownership; + } + + private void addOwnership(Token current, Token next, Map ownership) + { + double size = current.size(next); + Token representative = current.getPartitioner().midpoint(current, next); + for (InetAddressAndPort n : replicationStrategy.calculateNaturalReplicas(representative, tokenMetadata).endpoints()) + { + Double v = ownership.get(n); + ownership.put(n, v != null ? v + size : size); + } + } } - static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final AbstractReplicationStrategy rs, final InetAddressAndPort endpoint) + private StrategyAdapter getOrCreateStrategy(InetAddressAndPort endpoint) { - if (rs instanceof NetworkTopologyStrategy) - return getStrategy(tokenMetadata, (NetworkTopologyStrategy) rs, rs.snitch, endpoint); - if (rs instanceof SimpleStrategy) - return getStrategy(tokenMetadata, (SimpleStrategy) rs, endpoint); - throw new ConfigurationException("Token allocation does not support replication strategy " + rs.getClass().getSimpleName()); + String dc = replicationStrategy.snitch.getDatacenter(endpoint); + String rack = replicationStrategy.snitch.getRack(endpoint); + return getOrCreateStrategy(dc, rack); } - static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final SimpleStrategy rs, final InetAddressAndPort endpoint) + private StrategyAdapter getOrCreateStrategy(String dc, String rack) { - final int replicas = rs.getReplicationFactor().allReplicas; + return strategyByRackDc.computeIfAbsent(dc, k -> new HashMap<>()).computeIfAbsent(rack, k -> createStrategy(dc, rack)); + } + private StrategyAdapter createStrategy(String dc, String rack) + { + if (replicationStrategy instanceof NetworkTopologyStrategy) + return createStrategy(tokenMetadata, (NetworkTopologyStrategy) replicationStrategy, dc, rack); + if (replicationStrategy instanceof SimpleStrategy) + return createStrategy((SimpleStrategy) replicationStrategy); + throw new ConfigurationException("Token allocation does not support replication strategy " + replicationStrategy.getClass().getSimpleName()); + } + + private StrategyAdapter createStrategy(final SimpleStrategy rs) + { + return createStrategy(rs.snitch, null, null, rs.getReplicationFactor().allReplicas, false); + } + + private StrategyAdapter createStrategy(TokenMetadata tokenMetadata, NetworkTopologyStrategy strategy, String dc, String rack) + { + int replicas = strategy.getReplicationFactor(dc).allReplicas; + + Topology topology = tokenMetadata.getTopology(); + // if topology hasn't been setup yet for this dc+rack then treat it as a separate unit + int racks = topology.getDatacenterRacks().get(dc) != null && topology.getDatacenterRacks().get(dc).containsKey(rack) + ? topology.getDatacenterRacks().get(dc).asMap().size() + : 1; + + if (replicas <= 1) + { + // each node is treated as separate and replicates once + return createStrategy(strategy.snitch, dc, null, 1, false); + } + else if (racks == replicas) + { + // each node is treated as separate and replicates once, with separate allocation rings for each rack + return createStrategy(strategy.snitch, dc, rack, 1, false); + } + else if (racks > replicas) + { + // group by rack + return createStrategy(strategy.snitch, dc, null, replicas, true); + } + else if (racks == 1) + { + return createStrategy(strategy.snitch, dc, null, replicas, false); + } + + throw new ConfigurationException(String.format("Token allocation failed: the number of racks %d in datacenter %s is lower than its replication factor %d.", + racks, dc, replicas)); + } + + // a null dc will always return true for inAllocationRing(..) + // a null rack will return true for inAllocationRing(..) for all nodes in the same dc + private StrategyAdapter createStrategy(IEndpointSnitch snitch, String dc, String rack, int replicas, boolean groupByRack) + { return new StrategyAdapter() { @Override @@ -197,141 +288,15 @@ public class TokenAllocation @Override public Object getGroup(InetAddressAndPort unit) { - return unit; + return groupByRack ? snitch.getRack(unit) : unit; } @Override public boolean inAllocationRing(InetAddressAndPort other) { - return true; + return (dc == null || dc.equals(snitch.getDatacenter(other))) && (rack == null || rack.equals(snitch.getRack(other))); } }; } - - static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final NetworkTopologyStrategy rs, final IEndpointSnitch snitch, final InetAddressAndPort endpoint) - { - final String dc = snitch.getDatacenter(endpoint); - final int replicas = rs.getReplicationFactor(dc).allReplicas; - return getStrategy(tokenMetadata, replicas, snitch, endpoint); - } - - static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final int replicas, final InetAddressAndPort endpoint) - { - return getStrategy(tokenMetadata, replicas, DatabaseDescriptor.getEndpointSnitch(), endpoint); - } - - static StrategyAdapter getStrategy(final TokenMetadata tokenMetadata, final int replicas, final IEndpointSnitch snitch, final InetAddressAndPort endpoint) - { - final String dc = snitch.getDatacenter(endpoint); - if (replicas == 0 || replicas == 1) - { - // No replication, each node is treated as separate. - return new StrategyAdapter() - { - @Override - public int replicas() - { - return 1; - } - - @Override - public Object getGroup(InetAddressAndPort unit) - { - return unit; - } - - @Override - public boolean inAllocationRing(InetAddressAndPort other) - { - return dc.equals(snitch.getDatacenter(other)); - } - }; - } - - Topology topology = tokenMetadata.getTopology(); - - // if topology hasn't been setup yet for this endpoint+rack then treat it as a separate unit - int racks = topology.getDatacenterRacks().get(dc) != null && topology.getDatacenterRacks().get(dc).containsKey(snitch.getRack(endpoint)) - ? topology.getDatacenterRacks().get(dc).asMap().size() - : 1; - - if (racks > replicas) - { - return new StrategyAdapter() - { - @Override - public int replicas() - { - return replicas; - } - - @Override - public Object getGroup(InetAddressAndPort unit) - { - return snitch.getRack(unit); - } - - @Override - public boolean inAllocationRing(InetAddressAndPort other) - { - return dc.equals(snitch.getDatacenter(other)); - } - }; - } - else if (racks == replicas) - { - // When the number of racks is the same as the replication factor, everything must replicate exactly once - // in each rack. This is the same as having independent rings from each rack. - final String rack = snitch.getRack(endpoint); - return new StrategyAdapter() - { - @Override - public int replicas() - { - return 1; - } - - @Override - public Object getGroup(InetAddressAndPort unit) - { - return unit; - } - - @Override - public boolean inAllocationRing(InetAddressAndPort other) - { - return dc.equals(snitch.getDatacenter(other)) && rack.equals(snitch.getRack(other)); - } - }; - } - else if (racks == 1) - { - // One rack, each node treated as separate. - return new StrategyAdapter() - { - @Override - public int replicas() - { - return replicas; - } - - @Override - public Object getGroup(InetAddressAndPort unit) - { - return unit; - } - - @Override - public boolean inAllocationRing(InetAddressAndPort other) - { - return dc.equals(snitch.getDatacenter(other)); - } - }; - } - else - throw new ConfigurationException( - String.format("Token allocation failed: the number of racks %d in datacenter %s is lower than its replication factor %d.", - racks, dc, replicas)); - } } diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index 832960662b..d5aacd0386 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -53,7 +53,6 @@ public abstract class AbstractReplicationStrategy @VisibleForTesting final String keyspaceName; - private Keyspace keyspace; public final Map configOptions; private final TokenMetadata tokenMetadata; @@ -64,14 +63,12 @@ public abstract class AbstractReplicationStrategy protected AbstractReplicationStrategy(String keyspaceName, TokenMetadata tokenMetadata, IEndpointSnitch snitch, Map configOptions) { - assert keyspaceName != null; assert snitch != null; assert tokenMetadata != null; this.tokenMetadata = tokenMetadata; this.snitch = snitch; this.configOptions = configOptions == null ? Collections.emptyMap() : configOptions; this.keyspaceName = keyspaceName; - // lazy-initialize keyspace itself since we don't create them until after the replication strategies } private final Map cachedReplicas = new NonBlockingHashMap<>(); @@ -204,13 +201,6 @@ public abstract class AbstractReplicationStrategy return resultResponseHandler; } - private Keyspace getKeyspace() - { - if (keyspace == null) - keyspace = Keyspace.open(keyspaceName); - return keyspace; - } - /** * calculate the RF based on strategy_options. When overwriting, ensure that this get() * is FAST, as this is called often. diff --git a/src/java/org/apache/cassandra/locator/SimpleSnitch.java b/src/java/org/apache/cassandra/locator/SimpleSnitch.java index d605b6e9dd..f26375a1d7 100644 --- a/src/java/org/apache/cassandra/locator/SimpleSnitch.java +++ b/src/java/org/apache/cassandra/locator/SimpleSnitch.java @@ -24,6 +24,8 @@ package org.apache.cassandra.locator; */ public class SimpleSnitch extends AbstractEndpointSnitch { + public static final String DATA_CENTER_NAME = "datacenter1"; + public String getRack(InetAddressAndPort endpoint) { return "rack1"; @@ -31,7 +33,7 @@ public class SimpleSnitch extends AbstractEndpointSnitch public String getDatacenter(InetAddressAndPort endpoint) { - return "datacenter1"; + return DATA_CENTER_NAME; } @Override diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index ca8fd998e2..546f44ec96 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -24,12 +24,10 @@ import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.*; - -import org.apache.cassandra.config.CassandraRelevantProperties; -import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict; import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,6 +38,7 @@ import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.BiMultiValMap; import org.apache.cassandra.utils.Pair; @@ -111,12 +110,20 @@ public class TokenMetadata public TokenMetadata() { - this(SortedBiMultiValMap.create(), + this(SortedBiMultiValMap.create(), HashBiMap.create(), Topology.empty(), DatabaseDescriptor.getPartitioner()); } + public TokenMetadata(IEndpointSnitch snitch) + { + this(SortedBiMultiValMap.create(), + HashBiMap.create(), + Topology.builder(() -> snitch).build(), + DatabaseDescriptor.getPartitioner()); + } + private TokenMetadata(BiMultiValMap tokenToEndpointMap, BiMap endpointsMap, Topology topology, IPartitioner partitioner) { this.tokenToEndpointMap = tokenToEndpointMap; @@ -1364,6 +1371,7 @@ public class TokenMetadata private final ImmutableMap> dcRacks; /** reverse-lookup map for endpoint to current known dc/rack assignment */ private final ImmutableMap> currentLocations; + private final Supplier snitchSupplier; private Topology(Builder builder) { @@ -1375,6 +1383,7 @@ public class TokenMetadata this.dcRacks = dcRackBuilder.build(); this.currentLocations = ImmutableMap.copyOf(builder.currentLocations); + this.snitchSupplier = builder.snitchSupplier; } /** @@ -1406,14 +1415,14 @@ public class TokenMetadata return new Builder(this); } - static Builder builder() + static Builder builder(Supplier snitchSupplier) { - return new Builder(); + return new Builder(snitchSupplier); } static Topology empty() { - return builder().build(); + return builder(() -> DatabaseDescriptor.getEndpointSnitch()).build(); } private static class Builder @@ -1424,12 +1433,14 @@ public class TokenMetadata private final Map> dcRacks; /** reverse-lookup map for endpoint to current known dc/rack assignment */ private final Map> currentLocations; + private final Supplier snitchSupplier; - Builder() + Builder(Supplier snitchSupplier) { this.dcEndpoints = HashMultimap.create(); this.dcRacks = new HashMap<>(); this.currentLocations = new HashMap<>(); + this.snitchSupplier = snitchSupplier; } Builder(Topology from) @@ -1441,6 +1452,7 @@ public class TokenMetadata dcRacks.put(entry.getKey(), HashMultimap.create(entry.getValue())); this.currentLocations = new HashMap<>(from.currentLocations); + this.snitchSupplier = from.snitchSupplier; } /** @@ -1448,9 +1460,8 @@ public class TokenMetadata */ Builder addEndpoint(InetAddressAndPort ep) { - IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch(); - String dc = snitch.getDatacenter(ep); - String rack = snitch.getRack(ep); + String dc = snitchSupplier.get().getDatacenter(ep); + String rack = snitchSupplier.get().getRack(ep); Pair current = currentLocations.get(ep); if (current != null) { diff --git a/src/java/org/apache/cassandra/tools/GenerateTokens.java b/src/java/org/apache/cassandra/tools/GenerateTokens.java new file mode 100644 index 0000000000..c03a4d0ce7 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/GenerateTokens.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.tools; + +import java.util.Arrays; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.GnuParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; + +import org.slf4j.LoggerFactory; + +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.tokenallocator.OfflineTokenAllocator; +import org.apache.cassandra.dht.tokenallocator.TokenAllocation; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.OutputHandler; + +public class GenerateTokens +{ + private static final String RF = "rf"; + private static final String TOKENS = "tokens"; + private static final String NODES = "nodes"; + private static final String PARTITIONER = "partitioner"; + private static final String RACKS = "racks"; + private static final String VERBOSE = "verbose"; + + public static void main(String[] args) + { + Options options = null; + int rf = 0; + int tokens = 0; + int nodes = 0; + IPartitioner partitioner = null; + int[] racksDef = null; + + OutputHandler logger = null; + + try + { + // disable the summary statistics logging, since this is a command-line tool with dedicated output + ((Logger) LoggerFactory.getLogger(TokenAllocation.class)).setLevel(Level.ERROR); + + Util.initDatabaseDescriptor(); + options = getOptions(); + CommandLine cmd = parseCommandLine(args, options); + + rf = Integer.parseInt(cmd.getOptionValue(RF)); + tokens = Integer.parseInt(cmd.getOptionValue(TOKENS)); + nodes = Integer.parseInt(cmd.getOptionValue(NODES)); + logger = new OutputHandler.SystemOutput(cmd.hasOption(VERBOSE), true, true); + + partitioner = FBUtilities.newPartitioner(cmd.getOptionValue(PARTITIONER, Murmur3Partitioner.class.getSimpleName())); + racksDef = getRacks(cmd.getOptionValue(RACKS, cmd.getOptionValue(NODES))); + if (Arrays.stream(racksDef).sum() != nodes) + { + throw new AssertionError(String.format("The sum of nodes in each rack %s must equal total node count %s.", + cmd.getOptionValue(RACKS), + cmd.getOptionValue(NODES))); + } + } + catch (NumberFormatException e) + { + System.err.println("Invalid integer " + e.getMessage()); + System.out.println(); + printUsage(options); + System.exit(1); + } + catch (AssertionError | ConfigurationException | ParseException t) + { + System.err.println(t.getMessage()); + System.out.println(); + printUsage(options); + System.exit(1); + } + + try + { + logger.output(String.format("Generating tokens for %d nodes with %d vnodes each for replication factor %d and partitioner %s", + nodes, tokens, rf, partitioner.getClass().getSimpleName())); + + for (OfflineTokenAllocator.FakeNode node : OfflineTokenAllocator.allocate(rf, tokens, racksDef, logger, partitioner)) + logger.output(String.format("Node %d rack %d: %s", node.nodeId(), node.rackId(), node.tokens().toString())); + } + catch (Throwable t) + { + logger.warn("Error running tool.", t); + System.exit(1); + } + } + + private static int[] getRacks(String racksDef) + { + return Arrays.stream(racksDef.split(",")).mapToInt(Integer::parseInt).toArray(); + } + + private static CommandLine parseCommandLine(String[] args, Options options) throws ParseException + { + return new GnuParser().parse(options, args, false); + } + + private static Options getOptions() + { + Options options = new Options(); + options.addOption(requiredOption("n", NODES, true, "Number of nodes.")); + options.addOption(requiredOption("t", TOKENS, true, "Number of tokens/vnodes per node.")); + options.addOption(requiredOption(null, RF, true, "Replication factor.")); + options.addOption("p", PARTITIONER, true, "Database partitioner, either Murmur3Partitioner or RandomPartitioner."); + options.addOption(null, RACKS, true, + "Number of nodes per rack, separated by commas. Must add up to the total node count.\n" + + "For example, 'generatetokens -n 30 -t 8 --rf 3 --racks 10,10,10' will generate tokens for\n" + + "three racks of 10 nodes each."); + options.addOption("v", VERBOSE, false, "Verbose logging."); + return options; + } + + private static Option requiredOption(String shortOpt, String longOpt, boolean hasArg, String description) + { + Option option = new Option(shortOpt, longOpt, hasArg, description); + option.setRequired(true); + return option; + } + + public static void printUsage(Options options) + { + String usage = "generatetokens -n NODES -t TOKENS --rf REPLICATION_FACTOR"; + String header = "--\n" + + "Generates tokens for a datacenter with the given number of nodes using the token allocation algorithm.\n" + + "Options are:"; + new HelpFormatter().printHelp(usage, header, options, ""); + } +} + diff --git a/src/java/org/apache/cassandra/utils/OutputHandler.java b/src/java/org/apache/cassandra/utils/OutputHandler.java index 88c54cedd3..3b10d613c7 100644 --- a/src/java/org/apache/cassandra/utils/OutputHandler.java +++ b/src/java/org/apache/cassandra/utils/OutputHandler.java @@ -18,6 +18,8 @@ */ package org.apache.cassandra.utils; +import java.io.PrintStream; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -62,11 +64,18 @@ public interface OutputHandler { public final boolean debug; public final boolean printStack; + public final PrintStream warnOut; public SystemOutput(boolean debug, boolean printStack) + { + this(debug, printStack, false); + } + + public SystemOutput(boolean debug, boolean printStack, boolean logWarnToStdErr) { this.debug = debug; this.printStack = printStack; + this.warnOut = logWarnToStdErr ? System.err : System.out; } public void output(String msg) @@ -87,9 +96,9 @@ public interface OutputHandler public void warn(String msg, Throwable th) { - System.out.println("WARNING: " + msg); + warnOut.println("WARNING: " + msg); if (printStack && th != null) - th.printStackTrace(System.out); + th.printStackTrace(warnOut); } } } diff --git a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java index c5cce58a9b..44ea5a96e2 100644 --- a/test/unit/org/apache/cassandra/dht/BootStrapperTest.java +++ b/test/unit/org/apache/cassandra/dht/BootStrapperTest.java @@ -17,27 +17,14 @@ */ package org.apache.cassandra.dht; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.fail; -import org.junit.Assert; - -import java.net.InetAddress; import java.net.UnknownHostException; -import java.util.Collection; import java.util.List; -import java.util.Map; import java.util.Random; -import java.util.UUID; import com.google.common.base.Predicate; import com.google.common.base.Predicates; import com.google.common.collect.Lists; import com.google.common.collect.Multimap; - -import org.apache.cassandra.dht.RangeStreamer.FetchReplica; -import org.apache.commons.math3.stat.descriptive.SummaryStatistics; - import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; @@ -46,21 +33,21 @@ import org.junit.runner.RunWith; import org.apache.cassandra.OrderedJUnit4ClassRunner; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.locator.InetAddressAndPort; -import org.apache.cassandra.locator.Replica; -import org.apache.cassandra.schema.Schema; import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.dht.tokenallocator.TokenAllocation; +import org.apache.cassandra.dht.RangeStreamer.FetchReplica; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.gms.IFailureDetectionEventListener; import org.apache.cassandra.gms.IFailureDetector; -import org.apache.cassandra.locator.IEndpointSnitch; -import org.apache.cassandra.locator.RackInferringSnitch; +import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.TokenMetadata; -import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.locator.Replica; +import org.apache.cassandra.schema.Schema; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.streaming.StreamOperation; -import org.apache.cassandra.utils.FBUtilities; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + @RunWith(OrderedJUnit4ClassRunner.class) public class BootStrapperTest @@ -164,191 +151,9 @@ public class BootStrapperTest List tokens = Lists.newArrayListWithCapacity(numVNodes); for (int j = 0; j < numVNodes; ++j) tokens.add(p.getRandomToken(rand)); - + tmd.updateNormalTokens(tokens, addr); } } - - @Test - public void testAllocateTokens() throws UnknownHostException - { - int vn = 16; - String ks = "BootStrapperTestKeyspace3"; - TokenMetadata tm = new TokenMetadata(); - generateFakeEndpoints(tm, 10, vn); - InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort(); - allocateTokensForNode(vn, ks, tm, addr); - } - @Test - public void testAllocateTokensLocalRf() throws UnknownHostException - { - int vn = 16; - int allocateTokensForLocalRf = 3; - TokenMetadata tm = new TokenMetadata(); - generateFakeEndpoints(tm, 10, vn); - InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort(); - allocateTokensForNode(vn, allocateTokensForLocalRf, tm, addr); - } - - public void testAllocateTokensNetworkStrategy(int rackCount, int replicas) throws UnknownHostException - { - IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch(); - try - { - DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch()); - int vn = 16; - String ks = "BootStrapperTestNTSKeyspace" + rackCount + replicas; - String dc = "1"; - - // Register peers with expected DC for NetworkTopologyStrategy. - TokenMetadata metadata = StorageService.instance.getTokenMetadata(); - metadata.clearUnsafe(); - metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.1.0.99")); - metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.15.0.99")); - - SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1")); - TokenMetadata tm = StorageService.instance.getTokenMetadata(); - tm.clearUnsafe(); - for (int i = 0; i < rackCount; ++i) - generateFakeEndpoints(tm, 10, vn, dc, Integer.toString(i)); - InetAddressAndPort addr = InetAddressAndPort.getByName("127." + dc + ".0.99"); - allocateTokensForNode(vn, ks, tm, addr); - // Note: Not matching replication factor in second datacentre, but this should not affect us. - } finally { - DatabaseDescriptor.setEndpointSnitch(oldSnitch); - } - } - - @Test - public void testAllocateTokensNetworkStrategyOneRack() throws UnknownHostException - { - testAllocateTokensNetworkStrategy(1, 3); - } - - @Test(expected = ConfigurationException.class) - public void testAllocateTokensNetworkStrategyTwoRacks() throws UnknownHostException - { - testAllocateTokensNetworkStrategy(2, 3); - } - - @Test - public void testAllocateTokensNetworkStrategyThreeRacks() throws UnknownHostException - { - testAllocateTokensNetworkStrategy(3, 3); - } - - @Test - public void testAllocateTokensNetworkStrategyFiveRacks() throws UnknownHostException - { - testAllocateTokensNetworkStrategy(5, 3); - } - - @Test - public void testAllocateTokensNetworkStrategyOneRackOneReplica() throws UnknownHostException - { - testAllocateTokensNetworkStrategy(1, 1); - } - - private void allocateTokensForNode(int vn, String ks, TokenMetadata tm, InetAddressAndPort addr) - { - SummaryStatistics os = TokenAllocation.replicatedOwnershipStats(tm.cloneOnlyTokenMap(), Keyspace.open(ks).getReplicationStrategy(), addr); - Collection tokens = BootStrapper.allocateTokens(tm, addr, ks, vn, 0); - assertEquals(vn, tokens.size()); - tm.updateNormalTokens(tokens, addr); - SummaryStatistics ns = TokenAllocation.replicatedOwnershipStats(tm.cloneOnlyTokenMap(), Keyspace.open(ks).getReplicationStrategy(), addr); - verifyImprovement(os, ns); - } - - private void allocateTokensForNode(int vn, int rf, TokenMetadata tm, InetAddressAndPort addr) - { - Collection tokens = BootStrapper.allocateTokens(tm, addr, rf, vn, 0); - assertEquals(vn, tokens.size()); - tm.updateNormalTokens(tokens, addr); - // SummaryStatistics is not implemented for `allocate_tokens_for_local_replication_factor` so can't be verified - } - - private void verifyImprovement(SummaryStatistics os, SummaryStatistics ns) - { - if (ns.getStandardDeviation() > os.getStandardDeviation()) - { - fail(String.format("Token allocation unexpectedly increased standard deviation.\nStats before:\n%s\nStats after:\n%s", os, ns)); - } - } - - @Test - public void testAllocateTokensRfEqRacks() throws UnknownHostException - { - IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch(); - try - { - DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch()); - int vn = 8; - int replicas = 3; - int rackCount = replicas; - String ks = "BootStrapperTestNTSKeyspaceRfEqRacks"; - String dc = "1"; - - TokenMetadata metadata = StorageService.instance.getTokenMetadata(); - metadata.clearUnsafe(); - metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.1.0.99")); - metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.15.0.99")); - - SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1")); - int base = 5; - for (int i = 0; i < rackCount; ++i) - generateFakeEndpoints(metadata, base << i, vn, dc, Integer.toString(i)); // unbalanced racks - - int cnt = 5; - for (int i = 0; i < cnt; ++i) - allocateTokensForNode(vn, ks, metadata, InetAddressAndPort.getByName("127." + dc + ".0." + (99 + i))); - - double target = 1.0 / (base + cnt); - double permittedOver = 1.0 / (2 * vn + 1) + 0.01; - - Map ownership = StorageService.instance.effectiveOwnership(ks); - boolean failed = false; - for (Map.Entry o : ownership.entrySet()) - { - int rack = o.getKey().getAddress()[2]; - if (rack != 0) - continue; - - System.out.format("Node %s owns %f ratio to optimal %.2f\n", o.getKey(), o.getValue(), o.getValue() / target); - if (o.getValue()/target > 1 + permittedOver) - failed = true; - } - Assert.assertFalse(String.format("One of the nodes in the rack has over %.2f%% overutilization.", permittedOver * 100), failed); - } finally { - DatabaseDescriptor.setEndpointSnitch(oldSnitch); - } - } - - @Test - public void testAllocateTokensMultipleKeyspaces() throws UnknownHostException - { - // TODO: This scenario isn't supported very well. Investigate a multi-keyspace version of the algorithm. - int vn = 16; - String ks3 = "BootStrapperTestKeyspace4"; // RF = 3 - String ks2 = "BootStrapperTestKeyspace5"; // RF = 2 - - TokenMetadata tm = new TokenMetadata(); - generateFakeEndpoints(tm, 10, vn); - - InetAddressAndPort dcaddr = FBUtilities.getBroadcastAddressAndPort(); - SummaryStatistics os3 = TokenAllocation.replicatedOwnershipStats(tm, Keyspace.open(ks3).getReplicationStrategy(), dcaddr); - SummaryStatistics os2 = TokenAllocation.replicatedOwnershipStats(tm, Keyspace.open(ks2).getReplicationStrategy(), dcaddr); - String cks = ks3; - String nks = ks2; - for (int i=11; i<=20; ++i) - { - allocateTokensForNode(vn, cks, tm, InetAddressAndPort.getByName("127.0.0." + (i + 1))); - String t = cks; cks = nks; nks = t; - } - - SummaryStatistics ns3 = TokenAllocation.replicatedOwnershipStats(tm, Keyspace.open(ks3).getReplicationStrategy(), dcaddr); - SummaryStatistics ns2 = TokenAllocation.replicatedOwnershipStats(tm, Keyspace.open(ks2).getReplicationStrategy(), dcaddr); - verifyImprovement(os3, ns3); - verifyImprovement(os2, ns2); - } } diff --git a/test/unit/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocatorTest.java b/test/unit/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocatorTest.java new file mode 100644 index 0000000000..832d13dca8 --- /dev/null +++ b/test/unit/org/apache/cassandra/dht/tokenallocator/OfflineTokenAllocatorTest.java @@ -0,0 +1,222 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.dht.tokenallocator; + +import java.util.Collection; +import java.util.List; + +import com.google.common.collect.Lists; + +import org.assertj.core.api.Assertions; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.dht.ByteOrderedPartitioner; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.RandomPartitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.tools.Util; +import org.apache.cassandra.utils.OutputHandler; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; +import static org.apache.cassandra.dht.tokenallocator.OfflineTokenAllocator.allocate; + +public class OfflineTokenAllocatorTest +{ + private static final Logger logger = LoggerFactory.getLogger(OfflineTokenAllocatorTest.class); + private static final OutputHandler FAIL_ON_WARN_OUTPUT = new SystemOutputImpl(); + + @Before + public void setup() + { + Util.initDatabaseDescriptor(); + } + + @Test(expected = UnsupportedOperationException.class) + public void testUnsupportedPartitioner() + { + List nodes = allocate(3, 4, new int[]{1,1,1}, FAIL_ON_WARN_OUTPUT, ByteOrderedPartitioner.instance); + Assert.assertEquals(3, nodes.size()); + } + + /** + * Cycle through a matrix of valid ranges. + */ + @Test + public void testTokenGenerations() + { + for (int numTokens = 1; numTokens <= 16 ; ++numTokens) + { + for (int rf = 1; rf <=5; ++rf) + { + int nodeCount = 32; + for (int racks = 1; racks <= 10; ++racks) + { + int[] nodeToRack = makeRackCountArray(nodeCount, racks); + for (IPartitioner partitioner : new IPartitioner[] { Murmur3Partitioner.instance, RandomPartitioner.instance }) + { + logger.info("Testing offline token allocator for numTokens={}, rf={}, racks={}, nodeToRack={}, partitioner={}", + numTokens, rf, racks, nodeToRack, partitioner); + assertTokensAndNodeCount(numTokens, nodeCount, allocate(rf, + numTokens, + nodeToRack, + new SystemOutputImpl(rf, racks), + partitioner)); + } + } + } + } + } + + private void assertTokensAndNodeCount(int numTokens, int nodeCount, List nodes) + { + assertEquals(nodeCount, nodes.size()); + Collection allTokens = Lists.newArrayList(); + for (OfflineTokenAllocator.FakeNode node : nodes) + { + Assertions.assertThat(node.tokens()).hasSize(numTokens); + Assertions.assertThat(allTokens).doesNotContainAnyElementsOf(node.tokens()); + allTokens.addAll(node.tokens()); + } + } + + private static int[] makeRackCountArray(int nodes, int racks) + { + assert nodes > 0; + assert racks > 0; + // Distribute nodes among the racks in round-robin fashion in the order the user is supposed to start them. + int[] rackCounts = new int[racks]; + int rack = 0; + for (int node = 0; node < nodes; node++) + { + rackCounts[rack]++; + if (++rack == racks) + rack = 0; + } + return rackCounts; + } + + @Test(expected = IllegalArgumentException.class) + public void testTokenGenerator_more_rf_than_racks() + { + OfflineTokenAllocator.allocate(3, 16, new int[]{1, 1}, FAIL_ON_WARN_OUTPUT, Murmur3Partitioner.instance); + } + + + @Test(expected = IllegalArgumentException.class) + public void testTokenGenerator_more_rf_than_nodes() + { + OfflineTokenAllocator.allocate(3, 16, new int[]{2}, FAIL_ON_WARN_OUTPUT, Murmur3Partitioner.instance); + } + + @Test + public void testTokenGenerator_single_rack_or_single_rf() + { + int numTokens = 16; + // Simple cases, single rack or single replication. + assertTokensAndNodeCount(numTokens, 1, allocate(1, + numTokens, + new int[]{1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + assertTokensAndNodeCount(numTokens, 2, allocate(1, + numTokens, + new int[]{1, 1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + assertTokensAndNodeCount(numTokens, 2, allocate(1, + numTokens, + new int[]{2}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + } + + @Test + public void testTokenGenerator_unbalanced_racks() + { + int numTokens = 16; + // Simple cases, single rack or single replication. + assertTokensAndNodeCount(numTokens, 6, allocate(1, + numTokens, + new int[]{5, 1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + assertTokensAndNodeCount(numTokens, 7, allocate(1, + numTokens, + new int[]{5, 1, 1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + assertTokensAndNodeCount(numTokens, 6, allocate(3, + numTokens, + new int[]{5, 1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + assertTokensAndNodeCount(numTokens, 7, allocate(3, + numTokens, + new int[]{5, 1, 1}, + FAIL_ON_WARN_OUTPUT, + Murmur3Partitioner.instance)); + } + + private static class SystemOutputImpl extends OutputHandler.SystemOutput + { + private final int rf; + private final int racks; + + private SystemOutputImpl() + { + super(true, true); + rf = racks = 1; + } + + private SystemOutputImpl(int rf, int racks) + { + super(true, true); + this.rf = rf; + this.racks = racks; + } + + @Override + public void warn(String msg) + { + // We can only guarantee that ownership stdev won't increase above the warn threshold for racks==1 or racks==rf + if (racks == 1 || racks == rf) + fail(msg); + else + super.warn(msg); + } + + @Override + public void warn(String msg, Throwable th) + { + // We can only guarantee that ownership stdev won't increase above the warn threshold for racks==1 or racks==rf + if (racks == 1 || racks == rf) + fail(msg); + else + super.warn(msg, th); + } + } +} diff --git a/test/unit/org/apache/cassandra/dht/tokenallocator/TokenAllocationTest.java b/test/unit/org/apache/cassandra/dht/tokenallocator/TokenAllocationTest.java new file mode 100644 index 0000000000..ac372e4827 --- /dev/null +++ b/test/unit/org/apache/cassandra/dht/tokenallocator/TokenAllocationTest.java @@ -0,0 +1,310 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.dht.tokenallocator; + +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.UUID; + +import com.google.common.collect.Lists; +import org.apache.commons.math3.stat.descriptive.SummaryStatistics; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.IEndpointSnitch; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.locator.RackInferringSnitch; +import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.FBUtilities; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +public class TokenAllocationTest +{ + static IPartitioner oldPartitioner; + static Random rand = new Random(1); + + @BeforeClass + public static void setup() throws ConfigurationException + { + DatabaseDescriptor.daemonInitialization(); + oldPartitioner = StorageService.instance.setPartitionerUnsafe(Murmur3Partitioner.instance); + SchemaLoader.startGossiper(); + SchemaLoader.prepareServer(); + SchemaLoader.schemaDefinition("TokenAllocationTest"); + } + + @AfterClass + public static void tearDown() + { + DatabaseDescriptor.setPartitionerUnsafe(oldPartitioner); + } + + private static TokenAllocation createForTest(TokenMetadata tokenMetadata, int replicas, int numTokens) + { + return TokenAllocation.create(DatabaseDescriptor.getEndpointSnitch(), tokenMetadata, replicas, numTokens); + } + + @Test + public void testAllocateTokensForKeyspace() throws UnknownHostException + { + int vn = 16; + String ks = "TokenAllocationTestKeyspace3"; + TokenMetadata tm = new TokenMetadata(); + generateFakeEndpoints(tm, 10, vn); + InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort(); + allocateTokensForKeyspace(vn, ks, tm, addr); + } + + @Test + public void testAllocateTokensForLocalRF() throws UnknownHostException + { + int vn = 16; + int allocateTokensForLocalRf = 3; + TokenMetadata tm = new TokenMetadata(); + generateFakeEndpoints(tm, 10, vn); + InetAddressAndPort addr = FBUtilities.getBroadcastAddressAndPort(); + allocateTokensForLocalReplicationFactor(vn, allocateTokensForLocalRf, tm, addr); + } + + private Collection allocateTokensForKeyspace(int vnodes, String keyspace, TokenMetadata tm, InetAddressAndPort addr) + { + AbstractReplicationStrategy rs = Keyspace.open(keyspace).getReplicationStrategy(); + TokenAllocation tokenAllocation = TokenAllocation.create(tm, rs, vnodes); + return allocateAndVerify(vnodes, addr, tokenAllocation); + } + + private void allocateTokensForLocalReplicationFactor(int vnodes, int rf, TokenMetadata tm, InetAddressAndPort addr) + { + TokenAllocation tokenAllocation = createForTest(tm, rf, vnodes); + allocateAndVerify(vnodes, addr, tokenAllocation); + } + + private Collection allocateAndVerify(int vnodes, InetAddressAndPort addr, TokenAllocation tokenAllocation) + { + SummaryStatistics os = tokenAllocation.getAllocationRingOwnership(addr); + Collection tokens = tokenAllocation.allocate(addr); + assertEquals(vnodes, tokens.size()); + SummaryStatistics ns = tokenAllocation.getAllocationRingOwnership(addr); + if (ns.getStandardDeviation() > os.getStandardDeviation()) + { + fail(String.format("Token allocation unexpectedly increased standard deviation.\nStats before:\n%s\nStats after:\n%s", os, ns)); + } + return tokens; + } + + @Test + public void testAllocateTokensNetworkStrategyOneRack() throws UnknownHostException + { + testAllocateTokensNetworkStrategy(1, 3); + } + + @Test(expected = ConfigurationException.class) + public void testAllocateTokensNetworkStrategyTwoRacks() throws UnknownHostException + { + testAllocateTokensNetworkStrategy(2, 3); + } + + @Test + public void testAllocateTokensNetworkStrategyThreeRacks() throws UnknownHostException + { + testAllocateTokensNetworkStrategy(3, 3); + } + + @Test + public void testAllocateTokensNetworkStrategyFiveRacks() throws UnknownHostException + { + testAllocateTokensNetworkStrategy(5, 3); + } + + @Test + public void testAllocateTokensNetworkStrategyOneRackOneReplica() throws UnknownHostException + { + testAllocateTokensNetworkStrategy(1, 1); + } + + + public void testAllocateTokensNetworkStrategy(int rackCount, int replicas) throws UnknownHostException + { + IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch(); + try + { + DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch()); + int vn = 16; + String ks = "TokenAllocationTestNTSKeyspace" + rackCount + replicas; + String dc = "1"; + + // Register peers with expected DC for NetworkTopologyStrategy. + TokenMetadata metadata = StorageService.instance.getTokenMetadata(); + metadata.clearUnsafe(); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.1.0.99")); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.15.0.99")); + + SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1")); + TokenMetadata tm = StorageService.instance.getTokenMetadata(); + tm.clearUnsafe(); + for (int i = 0; i < rackCount; ++i) + generateFakeEndpoints(tm, 10, vn, dc, Integer.toString(i)); + InetAddressAndPort addr = InetAddressAndPort.getByName("127." + dc + ".0.99"); + allocateTokensForKeyspace(vn, ks, tm, addr); + // Note: Not matching replication factor in second datacentre, but this should not affect us. + } finally { + DatabaseDescriptor.setEndpointSnitch(oldSnitch); + } + } + + @Test + public void testAllocateTokensRfEqRacks() throws UnknownHostException + { + IEndpointSnitch oldSnitch = DatabaseDescriptor.getEndpointSnitch(); + try + { + DatabaseDescriptor.setEndpointSnitch(new RackInferringSnitch()); + int vn = 8; + int replicas = 3; + int rackCount = replicas; + String ks = "TokenAllocationTestNTSKeyspaceRfEqRacks"; + String dc = "1"; + + TokenMetadata metadata = StorageService.instance.getTokenMetadata(); + metadata.clearUnsafe(); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.1.0.99")); + metadata.updateHostId(UUID.randomUUID(), InetAddressAndPort.getByName("127.15.0.99")); + + SchemaLoader.createKeyspace(ks, KeyspaceParams.nts(dc, replicas, "15", 15), SchemaLoader.standardCFMD(ks, "Standard1")); + int base = 5; + for (int i = 0; i < rackCount; ++i) + generateFakeEndpoints(metadata, base << i, vn, dc, Integer.toString(i)); // unbalanced racks + + int cnt = 5; + for (int i = 0; i < cnt; ++i) + { + InetAddressAndPort endpoint = InetAddressAndPort.getByName("127." + dc + ".0." + (99 + i)); + Collection tokens = allocateTokensForKeyspace(vn, ks, metadata, endpoint); + metadata.updateNormalTokens(tokens, endpoint); + } + + double target = 1.0 / (base + cnt); + double permittedOver = 1.0 / (2 * vn + 1) + 0.01; + + Map ownership = StorageService.instance.effectiveOwnership(ks); + boolean failed = false; + for (Map.Entry o : ownership.entrySet()) + { + int rack = o.getKey().getAddress()[2]; + if (rack != 0) + continue; + + System.out.format("Node %s owns %f ratio to optimal %.2f\n", o.getKey(), o.getValue(), o.getValue() / target); + if (o.getValue()/target > 1 + permittedOver) + failed = true; + } + Assert.assertFalse(String.format("One of the nodes in the rack has over %.2f%% overutilization.", permittedOver * 100), failed); + } finally { + DatabaseDescriptor.setEndpointSnitch(oldSnitch); + } + } + + + /** + * TODO: This scenario isn't supported very well. Investigate a multi-keyspace version of the algorithm. + */ + @Test + public void testAllocateTokensMultipleKeyspaces() throws UnknownHostException + { + final int TOKENS = 16; + + TokenMetadata tokenMetadata = new TokenMetadata(); + generateFakeEndpoints(tokenMetadata, 10, TOKENS); + + // Do not clone token metadata so tokens allocated by different allocators are reflected on the parent TokenMetadata + TokenAllocation rf2Allocator = createForTest(tokenMetadata, 2, TOKENS); + TokenAllocation rf3Allocator = createForTest(tokenMetadata, 3, TOKENS); + + SummaryStatistics rf2StatsBefore = rf2Allocator.getAllocationRingOwnership(FBUtilities.getBroadcastAddressAndPort()); + SummaryStatistics rf3StatsBefore = rf3Allocator.getAllocationRingOwnership(FBUtilities.getBroadcastAddressAndPort()); + + TokenAllocation current = rf3Allocator; + TokenAllocation next = rf2Allocator; + + for (int i=11; i<=20; ++i) + { + InetAddressAndPort endpoint = InetAddressAndPort.getByName("127.0.0." + (i + 1)); + Collection tokens = current.allocate(endpoint); + // Update tokens on next to verify ownership calculation below + next.tokenMetadata.updateNormalTokens(tokens, endpoint); + TokenAllocation tmp = current; + current = next; + next = tmp; + } + + SummaryStatistics rf2StatsAfter = rf2Allocator.getAllocationRingOwnership(FBUtilities.getBroadcastAddressAndPort()); + SummaryStatistics rf3StatsAfter = rf3Allocator.getAllocationRingOwnership(FBUtilities.getBroadcastAddressAndPort()); + + verifyImprovement(rf2StatsBefore, rf2StatsAfter); + verifyImprovement(rf3StatsBefore, rf3StatsAfter); + } + + private void verifyImprovement(SummaryStatistics os, SummaryStatistics ns) + { + if (ns.getStandardDeviation() > os.getStandardDeviation()) + { + fail(String.format("Token allocation unexpectedly increased standard deviation.\nStats before:\n%s\nStats after:\n%s", os, ns)); + } + } + + private void generateFakeEndpoints(TokenMetadata tmd, int numOldNodes, int numVNodes) throws UnknownHostException + { + tmd.clearUnsafe(); + generateFakeEndpoints(tmd, numOldNodes, numVNodes, "0", "0"); + } + + private void generateFakeEndpoints(TokenMetadata tmd, int nodes, int vnodes, String dc, String rack) throws UnknownHostException + { + System.out.printf("Adding %d nodes to dc=%s, rack=%s.%n", nodes, dc, rack); + IPartitioner p = tmd.partitioner; + + for (int i = 1; i <= nodes; i++) + { + // leave .1 for myEndpoint + InetAddressAndPort addr = InetAddressAndPort.getByName("127." + dc + '.' + rack + '.' + (i + 1)); + List tokens = Lists.newArrayListWithCapacity(vnodes); + for (int j = 0; j < vnodes; ++j) + tokens.add(p.getRandomToken(rand)); + tmd.updateNormalTokens(tokens, addr); + } + } +} diff --git a/test/unit/org/apache/cassandra/tools/GenerateTokensTest.java b/test/unit/org/apache/cassandra/tools/GenerateTokensTest.java new file mode 100644 index 0000000000..0e90c85440 --- /dev/null +++ b/test/unit/org/apache/cassandra/tools/GenerateTokensTest.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.tools; + +import org.junit.Assert; +import org.junit.Test; + +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.dht.tokenallocator.TokenAllocation; + + +public class GenerateTokensTest +{ + + @Test + public void testMain() + { + for (int rf = 1; rf <= 5; ++rf) + { + for (int numTokens = 1; numTokens <= 16 ; ++numTokens) + { + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf)}); + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--partitioner", "Murmur3Partitioner"}); + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--partitioner", "RandomPartitioner"}); + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--racks", "15"}); + if (rf <= 2) + GenerateTokens.main(new String[]{"-n", "10", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--racks", "5,5"}); + if (rf <= 3) + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--racks", "5,5,5"}); + if (rf <= 4) + GenerateTokens.main(new String[]{"-n", "16", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--racks", "4,4,4,4"}); + if (rf == 5) + GenerateTokens.main(new String[]{"-n", "15", "-t", String.valueOf(numTokens), "--rf", String.valueOf(rf), "--racks", "3,3,3,3,3"}); + } + } + Assert.assertFalse(LoggerFactory.getLogger(TokenAllocation.class).isWarnEnabled()); + } + +} diff --git a/tools/bin/generatetokens b/tools/bin/generatetokens new file mode 100755 index 0000000000..e0a6b315cc --- /dev/null +++ b/tools/bin/generatetokens @@ -0,0 +1,54 @@ +#!/bin/sh + +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +if [ "x$CASSANDRA_INCLUDE" = "x" ]; then + # Locations (in order) to use when searching for an include file. + for include in "`dirname "$0"`/cassandra.in.sh" \ + "$HOME/.cassandra.in.sh" \ + /usr/share/cassandra/cassandra.in.sh \ + /usr/local/share/cassandra/cassandra.in.sh \ + /opt/cassandra/cassandra.in.sh; do + if [ -r "$include" ]; then + . "$include" + break + fi + done +elif [ -r "$CASSANDRA_INCLUDE" ]; then + . "$CASSANDRA_INCLUDE" +fi + +if [ -z "$CLASSPATH" ]; then + echo "You must set the CLASSPATH var" >&2 + exit 1 +fi + +if [ "x${MAX_HEAP_SIZE}" = "x" ]; then + MAX_HEAP_SIZE="256M" +fi + +if [ "x${MAX_DIRECT_MEMORY}" = "x" ]; then + MAX_DIRECT_MEMORY="2G" +fi + +JVM_OPTS="${JVM_OPTS} -Xmx${MAX_HEAP_SIZE} -XX:MaxDirectMemorySize=${MAX_DIRECT_MEMORY}" + +"${JAVA}" ${JAVA_AGENT} -ea -cp "${CLASSPATH}" ${JVM_OPTS} \ + -Dcassandra.storagedir="${cassandra_storagedir}" \ + -Dlogback.configurationFile=logback-tools.xml \ + org.apache.cassandra.tools.GenerateTokens "$@" +