From 171e1021fe6849af5575a145a52b40f6bf735ed7 Mon Sep 17 00:00:00 2001 From: Jun Rao Date: Sat, 5 Dec 2009 18:49:12 +0000 Subject: [PATCH] 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 --- .../cassandra/service/StorageService.java | 91 ++++++++++++++++--- .../service/StorageServiceMBean.java | 9 ++ .../org/apache/cassandra/tools/NodeProbe.java | 16 +++- 3 files changed, 100 insertions(+), 16 deletions(-) diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index d2aefb834a..7d6a8a33d5 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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 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() + { + 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 endpoints = getNaturalEndpoints(getLocalToken()); + foreachColumnFamily(new Function() + { + 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 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 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. diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index 3cb2001f50..b3f37abd47 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -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. */ diff --git a/src/java/org/apache/cassandra/tools/NodeProbe.java b/src/java/org/apache/cassandra/tools/NodeProbe.java index 347e2c16c2..a125a6638e 100644 --- a/src/java/org/apache/cassandra/tools/NodeProbe.java +++ b/src/java/org/apache/cassandra/tools/NodeProbe.java @@ -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 %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")) {