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

add manual repair through NodeProbe; patched by Stu Hood, reviewed by junrao for CASSANDRA-193 

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@887576 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jun Rao 2009-12-05 18:49:12 +00:00
parent ac398bd0a7
commit 171e1021fe
3 changed files with 100 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -37,6 +37,7 @@ import org.apache.cassandra.dht.*;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.locator.*;
import org.apache.cassandra.net.*;
import org.apache.cassandra.service.AntiEntropyService.TreeRequestVerbHandler;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.utils.FBUtilities;
@ -47,6 +48,7 @@ import org.apache.log4j.Logger;
import org.apache.log4j.Level;
import org.apache.commons.lang.StringUtils;
import com.google.common.base.Function;
import com.google.common.collect.Multimap;
import com.google.common.collect.HashMultimap;
@ -84,7 +86,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
private static volatile StorageService instance_;
public static IPartitioner<?> getPartitioner() {
public static IPartitioner getPartitioner() {
return partitioner_;
}
@ -276,6 +278,7 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
SelectorManager.getSelectorManager().start();
SelectorManager.getUdpSelectorManager().start();
AntiEntropyService.instance();
StorageLoadBalancer.instance().startBroadcasting();
// have to start the gossip service before we can see any info on other nodes. this is necessary
@ -686,12 +689,13 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
/**
* Flush all memtables for a table and column families.
* @param tableName
* @param columnFamilies
* Applies the given Function to all matching column families.
* @param function Function taking a column family and possibly returning an IOException.
* @param tableName Name of matching table.
* @param columnFamilies Names of matching column families, or null for all.
* @throws IOException
*/
public void forceTableFlush(String tableName, String... columnFamilies) throws IOException
public void foreachColumnFamily(Function<ColumnFamilyStore, IOException> function, String tableName, String... columnFamilies) throws IOException
{
if (DatabaseDescriptor.getTable(tableName) == null)
{
@ -709,14 +713,12 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
for (String columnFamily : columnFamilies)
{
if (positiveColumnFamilies.contains(columnFamily))
{
ColumnFamilyStore cfStore = table.getColumnFamilyStore(columnFamily);
logger_.debug("Forcing binary flush on keyspace " + tableName + ", CF " + columnFamily);
cfStore.forceFlushBinary();
logger_.debug("Forcing flush on keyspace " + tableName + ", CF " + columnFamily);
cfStore.forceFlush();
IOException result = function.apply(cfStore);
if (result != null)
throw result;
}
else
{
@ -726,6 +728,59 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
}
}
/**
* Flush all memtables for a table and column families.
* @param tableName
* @param columnFamilies
* @throws IOException
*/
public void forceTableFlush(final String tableName, final String... columnFamilies) throws IOException
{
foreachColumnFamily(new Function<ColumnFamilyStore, IOException>()
{
public IOException apply(ColumnFamilyStore cfStore)
{
try
{
logger_.debug("Forcing binary flush on keyspace " + tableName + ", CF " + cfStore.getColumnFamilyName());
cfStore.forceFlushBinary();
logger_.debug("Forcing flush on keyspace " + tableName + ", CF " + cfStore.getColumnFamilyName());
cfStore.forceFlush();
}
catch(IOException e)
{
return e;
}
return null;
}
}, tableName, columnFamilies);
}
/**
* Trigger proactive repair for a table and column families.
* @param tableName
* @param columnFamilies
* @throws IOException
*/
public void forceTableRepair(final String tableName, final String... columnFamilies) throws IOException
{
// request that all relevant endpoints generate trees
final MessagingService ms = MessagingService.instance();
final List<InetAddress> endpoints = getNaturalEndpoints(getLocalToken());
foreachColumnFamily(new Function<ColumnFamilyStore, IOException>()
{
public IOException apply(ColumnFamilyStore cfStore)
{
Message request = TreeRequestVerbHandler.makeVerb(tableName,
cfStore.getColumnFamilyName());
for (InetAddress endpoint : endpoints)
ms.sendOneWay(request, endpoint);
return null;
}
}, tableName, columnFamilies);
}
/* End of MBean interface methods */
/**
@ -852,9 +907,21 @@ public final class StorageService implements IEndPointStateChangeSubscriber, Sto
*/
public List<InetAddress> getNaturalEndpoints(String key)
{
return replicationStrategy_.getNaturalEndpoints(partitioner_.getToken(key));
}
return getNaturalEndpoints(partitioner_.getToken(key));
}
/**
* This method returns the N endpoints that are responsible for storing the
* specified key i.e for replication.
*
* @param token - token for which we need to find the endpoint return value -
* the endpoint responsible for this token
*/
public List<InetAddress> getNaturalEndpoints(Token token)
{
return replicationStrategy_.getNaturalEndpoints(token);
}
/**
* This method attempts to return N endpoints that are responsible for storing the
* specified key i.e for replication.

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

@ -114,6 +114,15 @@ public interface StorageServiceMBean
*/
public void forceTableFlush(String tableName, String... columnFamilies) throws IOException;
/**
* Triggers proactive repair for given column families, or all columnfamilies for the given table
* if none are explicitly listed.
* @param tableName
* @param columnFamilies
* @throws IOException
*/
public void forceTableRepair(String tableName, String... columnFamilies) throws IOException;
/**
* transfer this node's data to other machines and remove it from service.
*/

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

@ -186,6 +186,11 @@ public class NodeProbe
ssProxy.forceTableFlush(tableName, columnFamilies);
}
public void forceTableRepair(String tableName, String... columnFamilies) throws IOException
{
ssProxy.forceTableRepair(tableName, columnFamilies);
}
/**
* Write a textual representation of the Cassandra ring.
*
@ -493,7 +498,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, move, loadbalance, cancelpending, " +
"tpstats, flush, repair, decommission, move, loadbalance, cancelpending, " +
" getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])");
String usage = String.format("java %s -host <arg> <command>%n", NodeProbe.class.getName());
hf.printHelp(usage, "", options, header);
@ -589,7 +594,7 @@ public class NodeProbe
{
probe.printThreadPoolStats(System.out);
}
else if (cmdName.equals("flush"))
else if (cmdName.equals("flush") || cmdName.equals("repair"))
{
if (probe.getArgs().length < 2)
{
@ -602,8 +607,11 @@ public class NodeProbe
for (int i = 0; i < columnFamilies.length; i++)
{
columnFamilies[i] = probe.getArgs()[i + 2];
}
probe.forceTableFlush(probe.getArgs()[1], columnFamilies);
}
if (cmdName.equals("flush"))
probe.forceTableFlush(probe.getArgs()[1], columnFamilies);
else // cmdName.equals("repair")
probe.forceTableRepair(probe.getArgs()[1], columnFamilies);
}
else if (cmdName.equals("getcompactionthreshold"))
{