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

add Move command 

patch by jbellis; reviewed by Jaakko Laine for CASSANDRA-541

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@835890 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-11-13 15:57:31 +00:00
parent 6ede8c5d91
commit f3cdaa96b7
13 changed files with 115 additions and 240 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
bin/tokenupdater
View File

@ -1,49 +0,0 @@
#!/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
for include in /usr/share/cassandra/cassandra.in.sh \
/usr/local/share/cassandra/cassandra.in.sh \
/opt/cassandra/cassandra.in.sh \
`dirname $0`/cassandra.in.sh; do
if [ -r $include ]; then
. $include
break
fi
done
elif [ -r $CASSANDRA_INCLUDE ]; then
. $CASSANDRA_INCLUDE
fi
if [ -z $CASSANDRA_CONF -o -z $CLASSPATH ]; then
echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2
exit 1
fi
# Special-case path variables.
case "`uname`" in
CYGWIN*)
CLASSPATH=`cygpath -p -w "$CLASSPATH"`
CASSANDRA_CONF=`cygpath -p -w "$CASSANDRA_CONF"`
;;
esac
java -cp $CLASSPATH -Dstorage-config=$CASSANDRA_CONF \
org.apache.cassandra.tools.TokenUpdater $@
# vi:ai sw=4 ts=4 tw=0 et

2
src/java/org/apache/cassandra/db/SystemTable.java
View File

@ -127,7 +127,7 @@ public class SystemTable
if (initialToken == null)
token = p.getRandomToken();
else
token = p.getToken(initialToken);
token = p.getTokenFactory().fromString(initialToken);
logger.info("Saved Token not found. Using " + token);
// seconds-since-epoch isn't a foolproof new generation

2
src/java/org/apache/cassandra/dht/IPartitioner.java
View File

@ -72,6 +72,8 @@ public interface IPartitioner<T extends Token>
/**
* @return a Token that can be used to route a given key
* (This is NOT a method to create a Token from its string representation;
* for that, use TokenFactory.fromString.)
*/
public T getToken(String key);

1
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -50,6 +50,7 @@ public abstract class SSTable
{
private static final Logger logger = Logger.getLogger(SSTable.class);
public static final int FILES_ON_DISK = 3; // data, index, and bloom filter
protected String path;
protected IPartitioner partitioner;

3
src/java/org/apache/cassandra/io/Streaming.java
View File

@ -72,7 +72,7 @@ public class Streaming
if (sstables.isEmpty())
return;
StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[sstables.size()];
StreamContextManager.StreamContext[] streamContexts = new StreamContextManager.StreamContext[SSTable.FILES_ON_DISK * sstables.size()];
int i = 0;
for (SSTableReader sstable : sstables)
{
@ -130,7 +130,6 @@ public class Streaming
StreamContextManager.StreamStatus streamStatus = new StreamContextManager.StreamStatus(streamContext.getTargetFile(), streamContext.getExpectedBytes() );
String file = getNewFileNameFromOldContextAndNames(fileNames, streamContext);
//String file = DatabaseDescriptor.getDataFileLocationForTable(streamContext.getTable()) + File.separator + newFileName + "-Data.db";
if (logger.isDebugEnabled())
logger.debug("Received Data from : " + message.getFrom() + " " + streamContext.getTargetFile() + " " + file);
streamContext.setTargetFile(file);

1
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -88,6 +88,7 @@ public class TokenMetadata
lock.writeLock().lock();
try
{
tokenToEndPointMap.inverse().remove(endpoint);
if (!endpoint.equals(tokenToEndPointMap.put(token, endpoint)))
{
sortedTokens = sortTokens();

9
src/java/org/apache/cassandra/net/TcpConnection.java
View File

@ -71,7 +71,6 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
private TcpConnection(InetAddress from, InetAddress to, TcpConnectionManager pool, boolean streaming) throws IOException
{
logger_.debug("creating connection from " + from + " to " + to);
socketChannel_ = SocketChannel.open();
socketChannel_.socket().bind(new InetSocketAddress(from, 0));
socketChannel_.configureBlocking(false);
@ -291,18 +290,20 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable
void closeSocket()
{
logger_.warn("Closing down connection " + socketChannel_ + " with " + pendingWrites_.size() + " writes remaining.");
if (pendingWrites_.size() > 0)
logger_.error("Closing down connection " + socketChannel_ + " with " + pendingWrites_.size() + " writes remaining.");
cancel(key_);
pendingWrites_.clear();
}
void errorClose()
{
logger_.warn("Closing down connection " + socketChannel_);
logger_.info("Closing errored connection " + socketChannel_);
pendingWrites_.clear();
cancel(key_);
pendingWrites_.clear();
pool_.destroy(this);
if (pool_ != null)
pool_.destroy(this);
}
private void cancel(SelectionKey key)

4
src/java/org/apache/cassandra/net/io/StreamContextManager.java
View File

@ -261,7 +261,7 @@ public class StreamContextManager
{
List<StreamContext> context = ctxBag_.get(key);
if ( context == null )
throw new IllegalStateException("Streaming context has not been set.");
throw new IllegalStateException("Streaming context has not been set for " + key);
StreamContext streamContext = context.remove(0);
if ( context.isEmpty() )
ctxBag_.remove(key);
@ -272,7 +272,7 @@ public class StreamContextManager
{
List<StreamStatus> status = streamStatusBag_.get(key);
if ( status == null )
throw new IllegalStateException("Streaming status has not been set.");
throw new IllegalStateException("Streaming status has not been set for " + key);
StreamStatus streamStatus = status.remove(0);
if ( status.isEmpty() )
streamStatusBag_.remove(key);

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

@ -19,6 +19,7 @@
package org.apache.cassandra.service;
import java.io.IOException;
import java.io.IOError;
import java.lang.management.ManagementFactory;
import java.lang.reflect.Constructor;
import java.util.*;
@ -71,7 +72,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
/* All verb handler identifiers */
public final static String mutationVerbHandler_ = "ROW-MUTATION-VERB-HANDLER";
public final static String tokenVerbHandler_ = "TOKEN-VERB-HANDLER";
public final static String binaryVerbHandler_ = "BINARY-VERB-HANDLER";
public final static String readRepairVerbHandler_ = "READ-REPAIR-VERB-HANDLER";
public final static String readVerbHandler_ = "ROW-READ-VERB-HANDLER";
@ -165,7 +165,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
{
bootstrapSet.remove(s);
if (logger_.isDebugEnabled())
logger_.debug("Removed " + s + " as a bootstrap source");
logger_.debug("Removed " + s + " as a bootstrap source; remaining is [" + StringUtils.join(bootstrapSet, ", ") + "]");
if (bootstrapSet.isEmpty())
{
@ -222,7 +222,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
endPointSnitch_ = DatabaseDescriptor.getEndPointSnitch();
/* register the verb handlers */
MessagingService.instance().registerVerbHandlers(tokenVerbHandler_, new TokenUpdateVerbHandler());
MessagingService.instance().registerVerbHandlers(binaryVerbHandler_, new BinaryVerbHandler());
MessagingService.instance().registerVerbHandlers(mutationVerbHandler_, new RowMutationVerbHandler());
MessagingService.instance().registerVerbHandlers(readRepairVerbHandler_, new ReadRepairVerbHandler());
@ -258,8 +257,6 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
public void start() throws IOException
{
storageMetadata_ = SystemTable.initMetadata();
isBootstrapMode = DatabaseDescriptor.isAutoBootstrap()
&& !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped());
/* Listen for application messages */
MessagingService.instance().listen(FBUtilities.getLocalAddress());
@ -277,15 +274,14 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
Gossiper.instance().register(this);
Gossiper.instance().start(FBUtilities.getLocalAddress(), storageMetadata_.getGeneration());
if (isBootstrapMode)
if (DatabaseDescriptor.isAutoBootstrap()
&& !(DatabaseDescriptor.getSeeds().contains(FBUtilities.getLocalAddress()) || SystemTable.isBootstrapped()))
{
logger_.info("Starting in bootstrap mode (first, sleeping to get load information)");
StorageLoadBalancer.instance().waitForLoadInfo();
logger_.info("... got load info");
Token token = BootStrapper.getBootstrapToken(tokenMetadata_, StorageLoadBalancer.instance().getLoadInfo());
SystemTable.updateToken(token); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping
Gossiper.instance().addApplicationState(StorageService.STATE_BOOTSTRAPPING, new ApplicationState(partitioner_.getTokenFactory().toString(getLocalToken())));
new BootStrapper(replicationStrategy_, FBUtilities.getLocalAddress(), getLocalToken(), tokenMetadata_).startBootstrap(); // handles token update
startBootstrap(token);
// don't finish startup (enabling thrift) until after bootstrap is done
while (isBootstrapMode)
{
@ -303,13 +299,21 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
{
SystemTable.setBootstrapped(true);
Token token = storageMetadata_.getToken();
setToken(token);
tokenMetadata_.update(token, FBUtilities.getLocalAddress());
Gossiper.instance().addApplicationState(StorageService.STATE_NORMAL, new ApplicationState(partitioner_.getTokenFactory().toString(token)));
}
assert tokenMetadata_.sortedTokens().size() > 0;
}
private void startBootstrap(Token token) throws IOException
{
isBootstrapMode = true;
SystemTable.updateToken(token); // DON'T use setToken, that makes us part of the ring locally which is incorrect until we are done bootstrapping
Gossiper.instance().addApplicationState(StorageService.STATE_BOOTSTRAPPING, new ApplicationState(partitioner_.getTokenFactory().toString(token)));
new BootStrapper(replicationStrategy_, FBUtilities.getLocalAddress(), token, tokenMetadata_).startBootstrap(); // handles token update
}
public boolean isBootstrapMode()
{
return isBootstrapMode;
@ -935,7 +939,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
return tokens;
}
public void decommission()
public void decommission() throws InterruptedException
{
if (!tokenMetadata_.isMember(FBUtilities.getLocalAddress()))
throw new UnsupportedOperationException("local node is not a member of the token ring yet");
@ -947,15 +951,22 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
logger_.info("DECOMMISSIONING");
Gossiper.instance().addApplicationState(STATE_LEAVING, new ApplicationState(getLocalToken().toString()));
logger_.info("decommission sleeping " + Streaming.RING_DELAY);
try
{
Thread.sleep(Streaming.RING_DELAY);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
Thread.sleep(Streaming.RING_DELAY);
Runnable finishLeaving = new Runnable()
{
public void run()
{
Gossiper.instance().stop();
logger_.info("DECOMMISSION FINISHED.");
// let op be responsible for killing the process
}
};
unbootstrap(finishLeaving);
}
private void unbootstrap(final Runnable onFinish)
{
Multimap<Range, InetAddress> rangesMM = getChangedRangesForLeaving(FBUtilities.getLocalAddress());
if (logger_.isDebugEnabled())
logger_.debug("Ranges needing transfer are [" + StringUtils.join(rangesMM.keySet(), ",") + "]");
@ -970,7 +981,25 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
{
pending.remove(entry);
if (pending.isEmpty())
finishLeaving();
{
SystemTable.setBootstrapped(false);
tokenMetadata_.removeEndpoint(FBUtilities.getLocalAddress());
replicationStrategy_.removeObsoletePendingRanges();
if (logger_.isDebugEnabled())
logger_.debug("");
Gossiper.instance().addApplicationState(STATE_LEFT, new ApplicationState(getLocalToken().toString()));
try
{
Thread.sleep(2 * Gossiper.intervalInMillis_);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
onFinish.run();
}
}
};
StageManager.getStage(streamStage_).execute(new Runnable()
@ -984,6 +1013,35 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
}
public void move(String newToken) throws InterruptedException
{
if (tokenMetadata_.getPendingRanges(FBUtilities.getLocalAddress()).size() > 0)
throw new UnsupportedOperationException("data is currently moving to this node; unable to leave the ring");
final Token token = partitioner_.getTokenFactory().fromString(newToken); // make sure it's valid
logger_.info("moving to " + token);
Gossiper.instance().addApplicationState(STATE_LEAVING, new ApplicationState(getLocalToken().toString()));
logger_.info("move sleeping " + Streaming.RING_DELAY);
Thread.sleep(Streaming.RING_DELAY);
Runnable finishMoving = new Runnable()
{
public void run()
{
try
{
logger_.info("re-bootstrapping to new token " + token);
startBootstrap(token);
}
catch (IOException e)
{
throw new IOError(e);
}
}
};
unbootstrap(finishMoving);
}
public <T> QuorumResponseHandler<T> getResponseHandler(IResponseResolver<T> responseResolver, int blockFor, int consistency_level)
{
return replicationStrategy_.getResponseHandler(responseResolver, blockFor, consistency_level);
@ -994,20 +1052,4 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
return replicationStrategy_;
}
public void finishLeaving()
{
SystemTable.setBootstrapped(false);
Gossiper.instance().addApplicationState(STATE_LEFT, new ApplicationState(getLocalToken().toString()));
try
{
Thread.sleep(2 * Gossiper.intervalInMillis_);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
Gossiper.instance().stop();
logger_.info("DECOMMISSION FINISHED.");
// let op be responsible for killing the process
}
}

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

@ -23,6 +23,7 @@ import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.FutureTask;
import org.apache.cassandra.dht.Range;
import java.net.InetAddress;
@ -116,7 +117,13 @@ public interface StorageServiceMBean
/**
* transfer this node's data to other machines and remove it from service.
*/
public void decommission();
public void decommission() throws InterruptedException;
/**
* @param newToken token to move this node to.
* This node will unload its data onto its neighbors, and bootstrap to the new token.
*/
public void move(String newToken) throws InterruptedException;
/** set the logging level at runtime */
public void setLog4jLevel(String classQualifier, String level);

56
src/java/org/apache/cassandra/service/TokenUpdateVerbHandler.java
View File

@ -1,56 +0,0 @@
/**
* 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.service;
import java.io.IOException;
import org.apache.log4j.Logger;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.gms.ApplicationState;
public class TokenUpdateVerbHandler implements IVerbHandler
{
private static Logger logger_ = Logger.getLogger(TokenUpdateVerbHandler.class);
public void doVerb(Message message)
{
if (StorageService.instance().isBootstrapMode())
throw new UnsupportedOperationException("Cannot set token during bootstrap");
byte[] body = message.getMessageBody();
DataInputBuffer bufIn = new DataInputBuffer();
bufIn.reset(body, body.length);
try
{
Token token = Token.serializer().deserialize(bufIn);
StorageService.instance().setToken(token);
Gossiper.instance().addApplicationState(StorageService.STATE_NORMAL, new ApplicationState(StorageService.getPartitioner().getTokenFactory().toString(token)));
}
catch (IOException ex)
{
throw new RuntimeException(ex);
}
}
}

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

@ -380,11 +380,16 @@ public class NodeProbe
ssProxy.clearSnapshot();
}
public void decommission()
public void decommission() throws InterruptedException
{
ssProxy.decommission();
}
public void move(String newToken) throws InterruptedException
{
ssProxy.move(newToken);
}
/**
* Print out the size of the queues in the thread pools
*
@ -480,7 +485,7 @@ public class NodeProbe
HelpFormatter hf = new HelpFormatter();
String header = String.format(
"%nAvailable commands: ring, info, cleanup, compact, cfstats, snapshot [name], clearsnapshot, " +
"tpstats, flush, decommission, " +
"tpstats, flush, decommission, move, " +
" getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])");
String usage = String.format("java %s -host <arg> <command>%n", NodeProbe.class.getName());
hf.printHelp(usage, "", options, header);
@ -489,7 +494,7 @@ public class NodeProbe
/**
* @param args
*/
public static void main(String[] args) throws IOException
public static void main(String[] args) throws IOException, InterruptedException
{
NodeProbe probe = null;
try
@ -543,6 +548,14 @@ public class NodeProbe
{
probe.decommission();
}
else if (cmdName.equals("move"))
{
if (arguments.length <= 1)
{
System.err.println("missing token argument");
}
probe.move(arguments[1]);
}
else if (cmdName.equals("snapshot"))
{
String snapshotName = "";

86
src/java/org/apache/cassandra/tools/TokenUpdater.java
View File

@ -1,86 +0,0 @@
/**
* 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.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Token;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.SelectorManager;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.FileUtils;
public class TokenUpdater
{
private static final int port_ = 7000;
private static final long waitTime_ = 10000;
public static void main(String[] args) throws Throwable
{
if (args.length < 2)
{
System.out.println("Usage : java org.apache.cassandra.tools.TokenUpdater <ip:port> <token>");
System.exit(1);
}
Thread selectorThread = SelectorManager.getSelectorManager();
selectorThread.setDaemon(true);
selectorThread.start();
String ipPort = args[0];
IPartitioner p = StorageService.getPartitioner();
Token token = p.getTokenFactory().fromString(args[1]);
System.out.println("Partitioner is " + p.getClass() + ", token is: " + token);
System.out.println(p.getTokenFactory().getClass());
String[] ipPortPair = ipPort.split(":");
int port = 7000;
if (ipPortPair.length > 1)
{
port = Integer.valueOf(ipPortPair[1]);
}
InetSocketAddress target = new InetSocketAddress(ipPortPair[0], port);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
DataOutputStream dos = new DataOutputStream(bos);
Token.serializer().serialize(token, dos);
/* Construct the token update message to be sent */
Message tokenUpdateMessage = new Message(target.getAddress(),
"",
StorageService.tokenVerbHandler_,
bos.toByteArray());
System.out.println("Sending a token update message to " + target);
MessagingService.instance().sendOneWay(tokenUpdateMessage, target.getAddress());
Thread.sleep(TokenUpdater.waitTime_);
System.out.println("Done sending the update message");
MessagingService.shutdown();
FileUtils.shutdown();
}
}