natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Jonathan Ellis 2009-10-07 19:42:01 +00:00
parent ff3dd4dcec
commit 303951c458
5 changed files with 32 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
CHANGES.txt
View File

@ -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)

5
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -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;

79
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -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<EndPoint>();
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. * <i>nodes</i> 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<String> 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<String> tables = DatabaseDescriptor.getTables();

15
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -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.
*/

20
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -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 <arg> <command>%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);