From 303951c4584cc6db4e71f404e4622dc2d07acf54 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Wed, 7 Oct 2009 19:42:01 +0000 Subject: [PATCH] r/m nodeprobe bootstrap patch by jbellis; reviewed by Eric Evans for CASSANDRA-438 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@822877 13f79535-47bb-0310-9956-ffa450edef68 --- CHANGES.txt | 9 +++ .../org/apache/cassandra/gms/Gossiper.java | 5 ++ .../cassandra/service/StorageService.java | 79 ++++--------------- .../service/StorageServiceMBean.java | 15 +--- .../org/apache/cassandra/tools/NodeProbe.java | 20 +---- 5 files changed, 32 insertions(+), 96 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 7ccfba251f..d88cf84c70 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,12 @@ +0.5 dev + * All non-seed nodes will attempt to bootstrap when started, until + bootstrap successfully completes. -b option is removed. + * Unless a token is manually specified in the configuration xml, + a bootstraping node will use a token that gives it half the + keys from the most-heavily-loaded node in the cluster, + instead of generating a random token. + + 0.4.1 * Fix FlushPeriod columnfamily configuration regression (CASSANDRA-455) diff --git a/src/java/org/apache/cassandra/gms/Gossiper.java b/src/java/org/apache/cassandra/gms/Gossiper.java index a68ec30ea3..4d43e9f0c0 100644 --- a/src/java/org/apache/cassandra/gms/Gossiper.java +++ b/src/java/org/apache/cassandra/gms/Gossiper.java @@ -145,6 +145,7 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC StageManager.registerStage( Gossiper.GOSSIP_STAGE, new SingleThreadedStage("GMFD") ); } + /** Register with the Gossiper for EndPointState notifications */ public void register(IEndPointStateChangeSubscriber subscriber) { subscribers_.add(subscriber); @@ -881,6 +882,10 @@ public class Gossiper implements IFailureDetectionEventListener, IEndPointStateC } } + /** + * Start the gossiper with the generation # retrieved from the System + * table + */ public void start(EndPoint localEndPoint, int generationNbr) throws IOException { localEndPoint_ = localEndPoint; diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index c2f9ff21a2..8bf3e888ea 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -83,7 +83,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto public final static String rangeVerbHandler_ = "RANGE-VERB-HANDLER"; public final static String bootstrapTokenVerbHandler_ = "SPLITS-VERB-HANDLER"; - private static StorageService instance_; + private static volatile StorageService instance_; private static EndPoint tcpAddr_; private static EndPoint udpAddr_; private static IPartitioner partitioner_; @@ -139,15 +139,13 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto { if (instance_ == null) { - boolean bootstrap = !(DatabaseDescriptor.getSeeds().contains(getLocalControlEndPoint().getHost()) || SystemTable.isBootstrapped()); - synchronized (StorageService.class) { if (instance_ == null) { try { - instance_ = new StorageService(bootstrap); + instance_ = new StorageService(); } catch (Throwable th) { @@ -248,9 +246,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } - public StorageService(boolean isBootstrapMode) + public StorageService() { - this.isBootstrapMode = isBootstrapMode; bootstrapSet = new HashSet(); init(); storageLoadBalancer_ = new StorageLoadBalancer(this); @@ -316,6 +313,8 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto storageMetadata_ = SystemTable.initMetadata(); tcpAddr_ = new EndPoint(DatabaseDescriptor.getStoragePort()); udpAddr_ = new EndPoint(DatabaseDescriptor.getControlPort()); + isBootstrapMode = !(DatabaseDescriptor.getSeeds().contains(udpAddr_.getHost()) || SystemTable.isBootstrapped()); + /* Listen for application messages */ MessagingService.instance().listen(tcpAddr_); /* Listen for control messages */ @@ -327,19 +326,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto /* starts a load timer thread */ loadTimer_.schedule( new LoadDisseminator(), StorageService.threshold_, StorageService.threshold_); - /* Start the storage load balancer */ - storageLoadBalancer_.start(); - /* Register with the Gossiper for EndPointState notifications */ - Gossiper.instance().register(this); - /* - * Start the gossiper with the generation # retrieved from the System - * table - */ - Gossiper.instance().start(udpAddr_, storageMetadata_.getGeneration()); - /* Make sure this token gets gossiped around. */ - tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode); - ApplicationState state = new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(storageMetadata_.getToken())); - Gossiper.instance().addApplicationState(StorageService.nodeId_, state); if (isBootstrapMode) { logger_.info("Starting in bootstrap mode (first, sleeping to get load information)"); @@ -379,9 +365,19 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto updateToken(t); } } - doBootstrap(StorageService.getLocalStorageEndPoint()); + + BootStrapper bs = new BootStrapper(new EndPoint[] {getLocalStorageEndPoint()}, storageMetadata_.getToken()); + bootStrapper_.submit(bs); Gossiper.instance().addApplicationState(BOOTSTRAP_MODE, new ApplicationState("")); } + + storageLoadBalancer_.start(); + Gossiper.instance().register(this); + Gossiper.instance().start(udpAddr_, storageMetadata_.getGeneration()); + /* Make sure this token gets gossiped around. */ + tokenMetadata_.update(storageMetadata_.getToken(), StorageService.tcpAddr_, isBootstrapMode); + ApplicationState state = new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(storageMetadata_.getToken())); + Gossiper.instance().addApplicationState(StorageService.nodeId_, state); } private Token getBootstrapTokenFrom(EndPoint maxEndpoint) @@ -659,42 +655,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto updateToken(newToken); } } - - /** - * This method takes a colon separated string of nodes that need - * to be bootstrapped. * nodes must be specified as A:B:C. - * @throws UnknownHostException - * - */ - private void doBootstrap(String nodes) throws UnknownHostException - { - String[] allNodes = nodes.split(":"); - EndPoint[] endpoints = new EndPoint[allNodes.length]; - Token[] tokens = new Token[allNodes.length]; - - for ( int i = 0; i < allNodes.length; ++i ) - { - String host = allNodes[i].trim(); - InetAddress ip = InetAddress.getByName(host); - host = ip.getHostAddress(); - endpoints[i] = new EndPoint( host, DatabaseDescriptor.getStoragePort() ); - tokens[i] = tokenMetadata_.getToken(endpoints[i]); - } - - /* Start the bootstrap algorithm */ - bootStrapper_.submit( new BootStrapper(endpoints, tokens) ); - } - /** - * Starts the bootstrap operations for the specified endpoint. - * @param endpoint - */ - public final void doBootstrap(EndPoint endpoint) - { - Token token = tokenMetadata_.getToken(endpoint); - bootStrapper_.submit(new BootStrapper(new EndPoint[]{endpoint}, token)); - } - /** * Deliver hints to the specified node when it has crashed * and come back up/ marked as alive after a network partition @@ -757,11 +718,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto return Gossiper.instance().getCurrentGenerationNumber(udpAddr_); } - public void bootstrapNodes(String nodes) throws UnknownHostException - { - doBootstrap(nodes); - } - public void forceTableCleanup() throws IOException { List tables = DatabaseDescriptor.getTables(); @@ -772,9 +728,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto } } - /** - * Trigger the immediate compaction of all tables. - */ public void forceTableCompaction() throws IOException { List tables = DatabaseDescriptor.getTables(); diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index c8044e7e94..c3aa6e7e84 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -78,20 +78,7 @@ public interface StorageServiceMBean * Forces major compaction (all sstable files compacted) */ public void forceTableCompaction() throws IOException; - - /** - * This method will cause the local node initiate - * the bootstrap process for all the nodes specified - * in the string parameter passed in. This local node - * will calculate who gives what ranges to the nodes - * and then instructs the nodes to do so. - * - * @param nodes colon delimited list of endpoints that need - * to be bootstrapped - * @throws UnknownHostException - */ - public void bootstrapNodes(String nodes) throws UnknownHostException; - + /** * Trigger a cleanup of keys on all tables. */ diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 83fcd67f8b..a4eab962ab 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -178,11 +178,6 @@ public class NodeProbe ssProxy.forceTableCleanup(); } - public void bootstrapNodes(String nodeList) throws UnknownHostException - { - ssProxy.bootstrapNodes(nodeList); - } - public void forceTableCompaction() throws IOException { ssProxy.forceTableCompaction(); @@ -469,7 +464,7 @@ public class NodeProbe { HelpFormatter hf = new HelpFormatter(); String header = String.format( - "%nAvailable commands: ring, cluster, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, bootstrap, tpstats, flush_binary, " + + "%nAvailable commands: ring, cluster, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, tpstats, flush_binary, " + " getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])"); String usage = String.format("java %s -host %n", NodeProbe.class.getName()); hf.printHelp(usage, "", options, header); @@ -541,19 +536,6 @@ public class NodeProbe { probe.clearSnapshot(); } - else if (cmdName.equals("bootstrap")) - { - if (arguments.length == 2) - { - probe.bootstrapNodes(arguments[1]); - } - else - { - System.err.println(cmdName + " needs a node to work with"); - NodeProbe.printUsage(); - System.exit(1); - } - } else if (cmdName.equals("tpstats")) { probe.printThreadPoolStats(System.out);