p/4443/030_relocate_token Implement token/range relocation.
Patch by eevans; reviewed by Brandon Williams for CASSANDRA-4559 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/a02bbf50 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/a02bbf50 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/a02bbf50 Branch: refs/heads/trunk Commit: a02bbf50eb09f1364de21eb41d91e6521610092b Parents: 9f643a4 Author: Eric Evans <[email protected]> Authored: Fri Sep 14 10:08:58 2012 -0500 Committer: Eric Evans <[email protected]> Committed: Fri Sep 14 10:11:08 2012 -0500 ---------------------------------------------------------------------- src/java/org/apache/cassandra/db/SystemTable.java | 16 ++ .../org/apache/cassandra/gms/VersionedValue.java | 7 + .../apache/cassandra/locator/TokenMetadata.java | 10 + .../apache/cassandra/service/StorageService.java | 151 ++++++++++++++- .../cassandra/service/StorageServiceMBean.java | 7 + 5 files changed, 182 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/a02bbf50/src/java/org/apache/cassandra/db/SystemTable.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index 1dfd3ea..a41dafb 100644 --- a/src/java/org/apache/cassandra/db/SystemTable.java +++ b/src/java/org/apache/cassandra/db/SystemTable.java @@ -194,6 +194,22 @@ public class SystemTable forceBlockingFlush(LOCAL_CF); } + /** + * Convenience method to update the list of tokens in the local system table. + * + * @param addTokens tokens to add + * @param rmTokens tokens to remove + * @return the collection of persisted tokens + */ + public static synchronized Collection<Token> updateLocalTokens(Collection<Token> addTokens, Collection<Token> rmTokens) + { + Collection<Token> tokens = getSavedTokens(); + tokens.removeAll(rmTokens); + tokens.addAll(addTokens); + updateTokens(tokens); + return tokens; + } + /** Serialize a collection of tokens to bytes */ private static ByteBuffer serializeTokens(Collection<Token> tokens) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/a02bbf50/src/java/org/apache/cassandra/gms/VersionedValue.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/gms/VersionedValue.java b/src/java/org/apache/cassandra/gms/VersionedValue.java index 92c79eb..8436adc 100644 --- a/src/java/org/apache/cassandra/gms/VersionedValue.java +++ b/src/java/org/apache/cassandra/gms/VersionedValue.java @@ -64,6 +64,7 @@ public class VersionedValue implements Comparable<VersionedValue> public final static String STATUS_LEAVING = "LEAVING"; public final static String STATUS_LEFT = "LEFT"; public final static String STATUS_MOVING = "MOVING"; + public final static String STATUS_RELOCATING = "RELOCATING"; public final static String REMOVING_TOKEN = "removing"; public final static String REMOVED_TOKEN = "removed"; @@ -158,6 +159,12 @@ public class VersionedValue implements Comparable<VersionedValue> return new VersionedValue(VersionedValue.STATUS_MOVING + VersionedValue.DELIMITER + partitioner.getTokenFactory().toString(token)); } + public VersionedValue relocating(Collection<Token> srcTokens) + { + return new VersionedValue( + versionString(VersionedValue.STATUS_RELOCATING, StringUtils.join(srcTokens, VersionedValue.DELIMITER))); + } + public VersionedValue hostId(UUID hostId) { return new VersionedValue(hostId.toString()); http://git-wip-us.apache.org/repos/asf/cassandra/blob/a02bbf50/src/java/org/apache/cassandra/locator/TokenMetadata.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/locator/TokenMetadata.java b/src/java/org/apache/cassandra/locator/TokenMetadata.java index 0ecb125..5a38ef4 100644 --- a/src/java/org/apache/cassandra/locator/TokenMetadata.java +++ b/src/java/org/apache/cassandra/locator/TokenMetadata.java @@ -889,6 +889,16 @@ public class TokenMetadata return sb.toString(); } + public String printRelocatingRanges() + { + StringBuilder sb = new StringBuilder(); + + for (Map.Entry<Token, InetAddress> entry : relocatingTokens.entrySet()) + sb.append(String.format("%s:%s%n", entry.getKey(), entry.getValue())); + + return sb.toString(); + } + public void invalidateCaches() { for (AbstractReplicationStrategy subscriber : subscribers) http://git-wip-us.apache.org/repos/asf/cassandra/blob/a02bbf50/src/java/org/apache/cassandra/service/StorageService.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 07ee2bd..241015f 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -163,7 +163,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe /* the probability for tracing any particular request, 0 disables tracing and 1 enables for all */ private double tracingProbability = 0.0; - private static enum Mode { NORMAL, CLIENT, JOINING, LEAVING, DECOMMISSIONED, MOVING, DRAINING, DRAINED } + private static enum Mode { NORMAL, CLIENT, JOINING, LEAVING, DECOMMISSIONED, MOVING, DRAINING, DRAINED, RELOCATING } private Mode operationMode; private final MigrationManager migrationManager = new MigrationManager(); @@ -1081,6 +1081,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe * Other STATUS values that may be seen (possibly anywhere in the normal progression): * STATUS_MOVING,newtoken * set if node is currently moving to a new token in the ring + * STATUS_RELOCATING,srcToken,srcToken,srcToken,... + * set if the endpoint is in the process of relocating a token to itself * REMOVING_TOKEN,deadtoken * set if the node is dead and is being removed by its REMOVAL_COORDINATOR * REMOVED_TOKEN,deadtoken @@ -1112,6 +1114,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe handleStateLeft(endpoint, pieces); else if (moveName.equals(VersionedValue.STATUS_MOVING)) handleStateMoving(endpoint, pieces); + else if (moveName.equals(VersionedValue.STATUS_RELOCATING)) + handleStateRelocating(endpoint, pieces); } } @@ -1185,7 +1189,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe * @param endpoint node * @param pieces STATE_NORMAL,token */ - private void handleStateNormal(InetAddress endpoint, String[] pieces) + private void handleStateNormal(final InetAddress endpoint, String[] pieces) { assert pieces.length >= 2; @@ -1212,10 +1216,11 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Set<Token> tokensToUpdateInMetadata = new HashSet<Token>(); Set<Token> tokensToUpdateInSystemTable = new HashSet<Token>(); + Set<Token> localTokensToRemove = new HashSet<Token>(); Set<InetAddress> endpointsToRemove = new HashSet<InetAddress>(); Multimap<InetAddress, Token> epToTokenCopy = getTokenMetadata().getEndpointToTokenMapForReading(); - for (Token token : tokens) + for (final Token token : tokens) { // we don't want to update if this node is responsible for the token and it has a later startup time than endpoint. InetAddress currentOwner = tokenMetadata.getEndpoint(token); @@ -1232,6 +1237,33 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // no need to persist, token/ip did not change tokensToUpdateInMetadata.add(token); } + else if (tokenMetadata.isRelocating(token) && tokenMetadata.getRelocatingRanges().get(token).equals(endpoint)) + { + // Token was relocating, this is the bookkeeping that makes it official. + tokensToUpdateInMetadata.add(token); + if (!isClientMode) + tokensToUpdateInSystemTable.add(token); + + optionalTasks.schedule(new Runnable() + { + public void run() + { + logger.info("Removing RELOCATION state for {} {}", endpoint, token); + getTokenMetadata().removeFromRelocating(token, endpoint); + } + }, RING_DELAY, TimeUnit.MILLISECONDS); + + // We used to own this token; This token will need to be removed from system.local + if (currentOwner.equals(FBUtilities.getBroadcastAddress())) + localTokensToRemove.add(token); + + logger.info("Token {} relocated to {}", token, endpoint); + } + else if (tokenMetadata.isRelocating(token)) + { + logger.info("Token {} is relocating to {}, ignoring update from {}", + new Object[]{token, tokenMetadata.getRelocatingRanges().get(token), endpoint}); + } else if (Gossiper.instance.compareEndpointStartup(endpoint, currentOwner) > 0) { tokensToUpdateInMetadata.add(token); @@ -1249,6 +1281,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe currentOwner, token, endpoint)); + if (logger.isDebugEnabled()) + logger.debug("Relocating ranges: {}", tokenMetadata.printRelocatingRanges()); } else { @@ -1257,6 +1291,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe currentOwner, token, endpoint)); + if (logger.isDebugEnabled()) + logger.debug("Relocating ranges: {}", tokenMetadata.printRelocatingRanges()); } } @@ -1264,6 +1300,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe for (InetAddress ep : endpointsToRemove) Gossiper.instance.removeEndpoint(ep); SystemTable.updateTokens(endpoint, tokensToUpdateInSystemTable); + SystemTable.updateLocalTokens(Collections.<Token>emptyList(), localTokensToRemove); if (tokenMetadata.isMoving(endpoint)) // if endpoint was moving to a new token tokenMetadata.removeFromMoving(endpoint); @@ -1351,6 +1388,26 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe } /** + * Handle one or more ranges (tokens) moving from their respective endpoints, to another. + * + * @param endpoint the destination of the move + * @param pieces STATE_RELOCATING,token,token,... + */ + private void handleStateRelocating(InetAddress endpoint, String[] pieces) + { + assert pieces.length >= 2; + + List<Token> tokens = new ArrayList<Token>(pieces.length - 1); + for (String tStr : Arrays.copyOfRange(pieces, 1, pieces.length)) + tokens.add(getPartitioner().getTokenFactory().fromString(tStr)); + + logger.debug("Tokens {} are relocating to {}", tokens, endpoint); + tokenMetadata.addRelocatingTokens(tokens, endpoint); + + calculatePendingRanges(); + } + + /** * Handle notification that a node being actively removed from the ring via 'removetoken' * * @param endpoint node @@ -1482,10 +1539,10 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe BiMultiValMap<Token, InetAddress> bootstrapTokens = tm.getBootstrapTokens(); Set<InetAddress> leavingEndpoints = tm.getLeavingEndpoints(); - if (bootstrapTokens.isEmpty() && leavingEndpoints.isEmpty() && tm.getMovingEndpoints().isEmpty()) + if (bootstrapTokens.isEmpty() && leavingEndpoints.isEmpty() && tm.getMovingEndpoints().isEmpty() && tm.getRelocatingRanges().isEmpty()) { if (logger.isDebugEnabled()) - logger.debug("No bootstrapping, leaving or moving nodes -> empty pending ranges for {}", table); + logger.debug("No bootstrapping, leaving or moving nodes, and no relocating tokens -> empty pending ranges for {}", table); tm.setPendingRanges(table, pendingRanges); return; } @@ -2538,8 +2595,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.moving(newToken)); setMode(Mode.MOVING, String.format("Moving %s from %s to %s.", localAddress, getLocalTokens().iterator().next(), newToken), true); - RangeRelocator relocator = new RangeRelocator(Collections.singleton(newToken), tablesToProcess); - setMode(Mode.MOVING, String.format("Sleeping %s ms before start streaming/fetching ranges", RING_DELAY), true); try { @@ -2550,6 +2605,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe throw new RuntimeException("Sleep interrupted " + e.getMessage()); } + RangeRelocator relocator = new RangeRelocator(Collections.singleton(newToken), tablesToProcess); + if (relocator.streamsNeeded()) { setMode(Mode.MOVING, "fetching new ranges and streaming old ranges", true); @@ -2642,6 +2699,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe { Set<InetAddress> currentEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetaClone)); Set<InetAddress> newEndpoints = ImmutableSet.copyOf(strategy.calculateNaturalEndpoints(toStream.right, tokenMetaCloneAllSettled)); + logger.debug("Range:" + toStream + "Current endpoints: " + currentEndpoints + " New endpoints: " + newEndpoints); rangeWithEndpoints.putAll(toStream, Sets.difference(newEndpoints, currentEndpoints)); } @@ -2683,6 +2741,83 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe } } + public void relocate(Collection<String> srcTokens) throws ConfigurationException + { + List<Token> tokens = new ArrayList<Token>(srcTokens.size()); + for (String srcT : srcTokens) + { + getPartitioner().getTokenFactory().validate(srcT); + tokens.add(getPartitioner().getTokenFactory().fromString(srcT)); + } + relocateTokens(tokens); + } + + private void relocateTokens(Collection<Token> srcTokens) + { + assert srcTokens != null; + InetAddress localAddress = FBUtilities.getBroadcastAddress(); + Collection<Token> localTokens = getTokenMetadata().getTokens(localAddress); + Set<Token> tokens = new HashSet<Token>(srcTokens); + + for (Token srcT : tokens) + { + if (localTokens.contains(srcT)) + { + tokens.remove(srcT); + logger.warn("cannot move {}; source and destination match", srcT); + } + } + + if (tokens.size() < 1) + logger.warn("no valid token arguments specified; nothing to relocate"); + + Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.relocating(tokens)); + setMode(Mode.RELOCATING, String.format("relocating %s to %s", tokens, localAddress.getHostAddress()), true); + + List<String> tables = Schema.instance.getNonSystemTables(); + + setMode(Mode.RELOCATING, String.format("Sleeping %s ms before start streaming/fetching ranges", RING_DELAY), true); + try + { + Thread.sleep(RING_DELAY); + } + catch (InterruptedException e) + { + throw new RuntimeException("Sleep interrupted " + e.getMessage()); + } + + RangeRelocator relocator = new RangeRelocator(tokens, tables); + + if (relocator.streamsNeeded()) + { + setMode(Mode.RELOCATING, "fetching new ranges and streaming old ranges", true); + + relocator.logStreamsMap("[Relocate->STREAMING]"); + CountDownLatch streamLatch = relocator.streams(); + + relocator.logRequestsMap("[Relocate->FETCHING]"); + CountDownLatch fetchLatch = relocator.requests(); + + try + { + streamLatch.await(); + fetchLatch.await(); + } + catch (InterruptedException e) + { + throw new RuntimeException("Interrupted latch while waiting for stream/fetch ranges to finish: " + e.getMessage()); + } + } + else + setMode(Mode.RELOCATING, "no new ranges to stream/fetch", true); + + Collection<Token> currentTokens = SystemTable.updateLocalTokens(tokens, Collections.<Token>emptyList()); + tokenMetadata.updateNormalTokens(currentTokens, FBUtilities.getBroadcastAddress()); + Gossiper.instance.addLocalApplicationState(ApplicationState.TOKENS, valueFactory.tokens(currentTokens)); + Gossiper.instance.addLocalApplicationState(ApplicationState.STATUS, valueFactory.normal(currentTokens)); + setMode(Mode.NORMAL, false); + } + /** * Get the status of a token removal. */ @@ -3243,7 +3378,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // adding difference ranges to fetch from a ring toStream.addAll(r1.subtract(r2)); intersect = true; - break; } } if (!intersect) @@ -3262,7 +3396,6 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe // adding difference ranges to fetch from a ring toFetch.addAll(r2.subtract(r1)); intersect = true; - break; } } if (!intersect) http://git-wip-us.apache.org/repos/asf/cassandra/blob/a02bbf50/src/java/org/apache/cassandra/service/StorageServiceMBean.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 77e9843..375dacd 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.nio.ByteBuffer; +import java.util.Collection; import java.util.List; import java.util.Map; import java.util.concurrent.ExecutionException; @@ -293,6 +294,12 @@ public interface StorageServiceMBean public void move(String newToken) throws IOException, InterruptedException, ConfigurationException; /** + * @param srcTokens tokens to move to this node + * @throws ConfigurationException when passed an invalid token string + */ + public void relocate(Collection<String> srcTokens) throws ConfigurationException; + + /** * removeToken removes token (and all data associated with * enpoint that had it) from the ring */
