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

Add nodetool options to repair specific ranges. 

Patch by brandonwilliams, reviewed by yukim for CASSANDRA-5280
This commit is contained in:
Brandon Williams 2013-02-22 15:43:42 -06:00
parent 4c98854bb2
commit 00748405bc
6 changed files with 81 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,5 +1,6 @@
1.1.11
* cli: Add JMX authentication support (CASSANDRA-5080)
* nodetool: ability to repair specific range (CASSANDRA-5280)
1.1.10

2
src/java/org/apache/cassandra/service/AntiEntropyService.java
View File

@ -169,7 +169,7 @@ public class AntiEntropyService
throw new IllegalArgumentException("Requested range intersects a local range but is not fully contained in one; this would lead to imprecise repair");
}
}
if (rangeSuperSet == null || !replicaSets.containsKey(toRepair))
if (rangeSuperSet == null || !replicaSets.containsKey(rangeSuperSet))
return Collections.emptySet();
Set<InetAddress> neighbors = new HashSet<InetAddress>(replicaSets.get(rangeSuperSet));

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

@ -1853,12 +1853,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
}
public int forceRepairAsync(final String tableName, final boolean isSequential, final boolean primaryRange, final String... columnFamilies)
{
final Collection<Range<Token>> ranges = primaryRange ? Collections.singletonList(getLocalPrimaryRange()) : getLocalRanges(tableName);
return forceRepairAsync(tableName, isSequential, ranges, columnFamilies);
}
public int forceRepairAsync(final String tableName, final boolean isSequential, final Collection<Range<Token>> ranges, final String... columnFamilies)
{
if (Table.SYSTEM_TABLE.equals(tableName))
return 0;
final int cmd = nextRepairCommand.incrementAndGet();
final Collection<Range<Token>> ranges = primaryRange ? Collections.singletonList(getLocalPrimaryRange()) : getLocalRanges(tableName);
if (ranges.size() > 0)
{
new Thread(new WrappedRunnable()
@ -1872,7 +1877,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
List<AntiEntropyService.RepairFuture> futures = new ArrayList<AntiEntropyService.RepairFuture>(ranges.size());
for (Range<Token> range : ranges)
{
AntiEntropyService.RepairFuture future = forceTableRepair(range, tableName, isSequential, columnFamilies);
AntiEntropyService.RepairFuture future;
try
{
future = forceTableRepair(range, tableName, isSequential, columnFamilies);
}
catch (IllegalArgumentException e)
{
message = String.format("Repair session failed with error: %s", e);
sendNotification("repair", message, new int[]{cmd, AntiEntropyService.Status.SESSION_FAILED.ordinal()});
continue;
}
if (future == null)
continue;
futures.add(future);
@ -1914,6 +1929,17 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
return cmd;
}
public int forceRepairRangeAsync(String beginToken, String endToken, final String tableName, boolean isSequential, final String... columnFamilies)
{
Token parsedBeginToken = getPartitioner().getTokenFactory().fromString(beginToken);
Token parsedEndToken = getPartitioner().getTokenFactory().fromString(endToken);
logger_.info("starting user-requested repair of range ({}, {}] for keyspace {} and column families {}",
new Object[] {parsedBeginToken, parsedEndToken, tableName, columnFamilies});
return forceRepairAsync(tableName, isSequential, Collections.singleton(new Range<Token>(parsedBeginToken, parsedEndToken)), columnFamilies);
}
/**
* Trigger proactive repair for a table and column families.
* @param tableName

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

@ -255,6 +255,11 @@ public interface StorageServiceMBean extends NotificationEmitter
*/
public int forceRepairAsync(String tableName, boolean isSequential, boolean primaryRange, String... columnFamilies);
/**
* Same as forceRepairAsync, but handles a specified range
*/
public int forceRepairRangeAsync(String beginToken, String endToken, final String tableName, boolean isSequential, final String... columnFamilies);
/**
* Triggers proactive repair for given column families, or all columnfamilies for the given table
* if none are explicitly listed.

9
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -55,6 +55,8 @@ public class NodeCmd
private static final Pair<String, String> PASSWORD_OPT = new Pair<String, String>("pw", "password");
private static final Pair<String, String> TAG_OPT = new Pair<String, String>("t", "tag");
private static final Pair<String, String> PRIMARY_RANGE_OPT = new Pair<String, String>("pr", "partitioner-range");
private static final Pair<String, String> START_TOKEN_OPT = new Pair<String, String>("st", "start-token");
private static final Pair<String, String> END_TOKEN_OPT = new Pair<String, String>("et", "end-token");
private static final Pair<String, String> SNAPSHOT_REPAIR_OPT = new Pair<String, String>("snapshot", "with-snapshot");
private static final String DEFAULT_HOST = "127.0.0.1";
@ -76,6 +78,8 @@ public class NodeCmd
options.addOption(TAG_OPT, true, "optional name to give a snapshot");
options.addOption(PRIMARY_RANGE_OPT, false, "only repair the first range returned by the partitioner for the node");
options.addOption(SNAPSHOT_REPAIR_OPT, false, "repair one node at a time using snapshots");
options.addOption(START_TOKEN_OPT, true, "token at which repair range starts");
options.addOption(END_TOKEN_OPT, true, "token at which repair range ends");
}
public NodeCmd(NodeProbe probe)
@ -1041,7 +1045,10 @@ public class NodeCmd
case REPAIR :
boolean snapshot = cmd.hasOption(SNAPSHOT_REPAIR_OPT.left);
boolean primaryRange = cmd.hasOption(PRIMARY_RANGE_OPT.left);
probe.forceRepairAsync(System.out, keyspace, snapshot, primaryRange, columnFamilies);
if (cmd.hasOption(START_TOKEN_OPT.left) || cmd.hasOption(END_TOKEN_OPT.left))
probe.forceRepairRangeAsync(System.out, keyspace, snapshot, cmd.getOptionValue(START_TOKEN_OPT.left), cmd.getOptionValue(END_TOKEN_OPT.left), columnFamilies);
else
probe.forceRepairAsync(System.out, keyspace, snapshot, primaryRange, columnFamilies);
break;
case FLUSH :
try { probe.forceTableFlush(keyspace, columnFamilies); }

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

@ -231,6 +231,29 @@ public class NodeProbe
}
}
public void forceRepairRangeAsync(final PrintStream out, final String tableName, boolean isSequential, final String startToken, final String endToken, String... columnFamilies) throws IOException
{
RepairRunner runner = new RepairRunner(out, tableName, columnFamilies);
try
{
ssProxy.addNotificationListener(runner, null, null);
if (!runner.repairRangeAndWait(ssProxy, isSequential, startToken, endToken))
failed = true;
}
catch (Exception e)
{
throw new IOException(e) ;
}
finally
{
try
{
ssProxy.removeNotificationListener(runner);
}
catch (ListenerNotFoundException ignored) {}
}
}
public void forceTableRepairPrimaryRange(String tableName, boolean isSequential, String... columnFamilies) throws IOException
{
ssProxy.forceTableRepairPrimaryRange(tableName, isSequential, columnFamilies);
@ -835,6 +858,21 @@ class RepairRunner implements NotificationListener
return success;
}
public boolean repairRangeAndWait(StorageServiceMBean ssProxy, boolean isSequential, String startToken, String endToken) throws InterruptedException
{
cmd = ssProxy.forceRepairRangeAsync(startToken, endToken, keyspace, isSequential, columnFamilies);
if (cmd > 0)
{
condition.await();
}
else
{
String message = String.format("[%s] Nothing to repair for keyspace '%s'", format.format(System.currentTimeMillis()), keyspace);
out.println(message);
}
return success;
}
public void handleNotification(Notification notification, Object handback)
{
if ("repair".equals(notification.getType()))