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

When lost notifications occur and periodically, check for the parent repair status and exit if we've completed/failed 

Patch by Matt Byrd, reviewed by Chris Lohfink for CASSANDRA-13480
This commit is contained in:
Matt Byrd 2017-05-26 17:03:17 -07:00 committed by Jeff Jirsa
parent e6ba616297
commit 20d5ce8b9b
8 changed files with 214 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -79,6 +79,7 @@
* Fix cqlsh automatic protocol downgrade regression (CASSANDRA-13307)
* Changing `max_hint_window_in_ms` at runtime (CASSANDRA-11720)
* Trivial format error in StorageProxy (CASSANDRA-13551)
* Nodetool repair can hang forever if we lose the notification for the repair completing/failing (CASSANDRA-13480)
3.11.1

115
src/java/org/apache/cassandra/repair/RepairRunnable.java
View File

@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
@ -109,7 +110,8 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
listeners.remove(listener);
}
protected void fireProgressEvent(String tag, ProgressEvent event)
protected void fireProgressEvent(ProgressEvent event)
{
for (ProgressListener listener : listeners)
{
@ -117,14 +119,17 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
}
protected void fireErrorAndComplete(String tag, int progressCount, int totalProgress, String message)
protected void fireErrorAndComplete(int progressCount, int totalProgress, String message)
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progressCount, totalProgress, message));
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.COMPLETE, progressCount, totalProgress, String.format("Repair command #%d finished with error", cmd)));
fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progressCount, totalProgress, message));
String completionMessage = String.format("Repair command #%d finished with error", cmd);
fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progressCount, totalProgress, completionMessage));
recordFailure(message, completionMessage);
}
protected void runMayThrow() throws Exception
{
ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.IN_PROGRESS, ImmutableList.of());
final TraceState traceState;
final UUID parentSession = UUIDGen.getTimeUUID();
final String tag = "repair:" + cmd;
@ -142,7 +147,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
catch (IllegalArgumentException e)
{
logger.error("Repair failed:", e);
fireErrorAndComplete(tag, progress.get(), totalProgress, e.getMessage());
fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
return;
}
@ -160,7 +165,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
traceState = Tracing.instance.begin("repair", ImmutableMap.of("keyspace", keyspace, "columnFamilies",
cfsb.substring(2)));
message = message + " tracing with " + sessionId;
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.START, 0, 100, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.START, 0, 100, message));
Tracing.traceRepair(message);
traceState.enableActivityNotification(tag);
for (ProgressListener listener : listeners)
@ -171,7 +176,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
else
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.START, 0, 100, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.START, 0, 100, message));
traceState = null;
}
@ -199,7 +204,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
catch (IllegalArgumentException e)
{
logger.error("Repair failed:", e);
fireErrorAndComplete(tag, progress.get(), totalProgress, e.getMessage());
fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
return;
}
@ -212,7 +217,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
catch (IllegalArgumentException e)
{
fireErrorAndComplete(tag, progress.get(), totalProgress, e.getMessage());
fireErrorAndComplete(progress.get(), totalProgress, e.getMessage());
return;
}
@ -240,7 +245,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
SystemDistributedKeyspace.failParentRepair(parentSession, t);
}
fireErrorAndComplete(tag, progress.get(), totalProgress, t.getMessage());
fireErrorAndComplete(progress.get(), totalProgress, t.getMessage());
return;
}
@ -352,22 +357,27 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
SyncStatSummary summary = new SyncStatSummary(true);
summary.consumeSessionResults(results);
final String message;
if (summary.isEmpty())
{
String message = previewKind == PreviewKind.REPAIRED ? "Repaired data is in sync" : "Previewed data was in sync";
message = previewKind == PreviewKind.REPAIRED ? "Repaired data is in sync" : "Previewed data was in sync";
logger.info(message);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
}
else
{
String message = (previewKind == PreviewKind.REPAIRED ? "Repaired data is inconsistent\n" : "Preview complete\n") + summary.toString();
message = (previewKind == PreviewKind.REPAIRED ? "Repaired data is inconsistent\n" : "Preview complete\n") + summary.toString();
logger.info(message);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, progress.get(), totalProgress, message));
}
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
"Repair preview completed successfully"));
complete();
String successMessage = "Repair preview completed successfully";
fireProgressEvent(new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
successMessage));
String completionMessage = complete();
ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.COMPLETED,
ImmutableList.of(message, successMessage, completionMessage));
}
catch (Throwable t)
{
@ -378,20 +388,22 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
public void onFailure(Throwable t)
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
logger.error("Error completing preview repair", t);
complete();
String completionMessage = complete();
recordFailure(t.getMessage(), completionMessage);
}
private void complete()
private String complete()
{
logger.debug("Preview repair {} completed", parentSession);
String duration = DurationFormatUtils.formatDurationWords(System.currentTimeMillis() - startTime,
true, true);
String message = String.format("Repair preview #%d finished in %s", cmd, duration);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
executor.shutdownNow();
return message;
}
});
}
@ -447,10 +459,10 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
String message = String.format("Repair session %s for range %s finished", session.getId(),
session.getRanges().toString());
logger.info(message);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.PROGRESS,
progress.incrementAndGet(),
totalProgress,
message));
fireProgressEvent(new ProgressEvent(ProgressEventType.PROGRESS,
progress.incrementAndGet(),
totalProgress,
message));
}
public void onFailure(Throwable t)
@ -458,10 +470,10 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
String message = String.format("Repair session %s for range %s failed with error %s",
session.getId(), session.getRanges().toString(), t.getMessage());
logger.error(message, t);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR,
progress.incrementAndGet(),
totalProgress,
message));
fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR,
progress.incrementAndGet(),
totalProgress,
message));
}
}
@ -495,36 +507,49 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
{
SystemDistributedKeyspace.successfulParentRepair(parentSession, successfulRanges);
}
final String message;
if (hasFailure.get())
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress,
"Some repair failed"));
message = "Some repair failed";
fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress,
message));
}
else
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
"Repair completed successfully"));
message = "Repair completed successfully";
fireProgressEvent(new ProgressEvent(ProgressEventType.SUCCESS, progress.get(), totalProgress,
message));
}
String completionMessage = repairComplete();
if (hasFailure.get())
{
recordFailure(message, completionMessage);
}
else
{
ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.COMPLETED,
ImmutableList.of(message, completionMessage));
}
repairComplete();
}
public void onFailure(Throwable t)
{
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
fireProgressEvent(new ProgressEvent(ProgressEventType.ERROR, progress.get(), totalProgress, t.getMessage()));
if (!options.isPreview())
{
SystemDistributedKeyspace.failParentRepair(parentSession, t);
}
repairComplete();
String completionMessage = repairComplete();
recordFailure(t.getMessage(), completionMessage);
}
private void repairComplete()
private String repairComplete()
{
ActiveRepairService.instance.removeParentRepairSession(parentSession);
long durationMillis = System.currentTimeMillis() - startTime;
String duration = DurationFormatUtils.formatDurationWords(durationMillis, true, true);
String duration = DurationFormatUtils.formatDurationWords(durationMillis,true, true);
String message = String.format("Repair command #%d finished in %s", cmd, duration);
fireProgressEvent(tag, new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.COMPLETE, progress.get(), totalProgress, message));
logger.info(message);
if (options.isTraced() && traceState != null)
{
@ -541,9 +566,18 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
}
executor.shutdownNow();
Keyspace.open(keyspace).metric.repairTime.update(durationMillis, TimeUnit.MILLISECONDS);
return message;
}
}
private void recordFailure(String failureMessage, String completionMessage)
{
// Note we rely on the first message being the reason for the failure
// when inspecting this state from RepairRunner.queryForCompletedRepair
ActiveRepairService.instance.recordRepairStatus(cmd, ActiveRepairService.ParentRepairStatus.FAILED,
ImmutableList.of(failureMessage, completionMessage));
}
private void addRangeToNeighbors(List<Pair<Set<InetAddress>, ? extends Collection<Range<Token>>>> neighborRangeList, Range<Token> range, Set<InetAddress> neighbors)
{
for (int i = 0; i < neighborRangeList.size(); i++)
@ -622,8 +656,7 @@ public class RepairRunnable extends WrappedRunnable implements ProgressEventNoti
if (seen[si == 0 ? 1 : 0].contains(uuid))
continue;
String message = String.format("%s: %s", r.getInetAddress("source"), r.getString("activity"));
fireProgressEvent("repair:" + cmd,
new ProgressEvent(ProgressEventType.NOTIFICATION, 0, 0, message));
fireProgressEvent(new ProgressEvent(ProgressEventType.NOTIFICATION, 0, 0, message));
}
tlast = tcur;

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

@ -29,6 +29,8 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.base.Predicate;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
@ -72,6 +74,7 @@ import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDGen;
/**
@ -90,6 +93,12 @@ import org.apache.cassandra.utils.UUIDGen;
*/
public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFailureDetectionEventListener, ActiveRepairServiceMBean
{
public enum ParentRepairStatus
{
IN_PROGRESS, COMPLETED, FAILED
}
public static class ConsistentSessions
{
public final LocalSessions local = new LocalSessions();
@ -118,11 +127,21 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
private final IFailureDetector failureDetector;
private final Gossiper gossiper;
private final Cache<Integer, Pair<ParentRepairStatus, List<String>>> repairStatusByCmd;
public ActiveRepairService(IFailureDetector failureDetector, Gossiper gossiper)
{
this.failureDetector = failureDetector;
this.gossiper = gossiper;
this.repairStatusByCmd = CacheBuilder.newBuilder()
.expireAfterWrite(
Long.getLong("cassandra.parent_repair_status_expiry_seconds",
TimeUnit.SECONDS.convert(1, TimeUnit.DAYS)), TimeUnit.SECONDS)
// using weight wouldn't work so well, since it doesn't reflect mutation of cached data
// see https://github.com/google/guava/wiki/CachesExplained
// We assume each entry is unlikely to be much more than 100 bytes, so bounding the size should be sufficient.
.maximumSize(Long.getLong("cassandra.parent_repair_status_cache_size", 100_000))
.build();
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
@ -230,6 +249,17 @@ public class ActiveRepairService implements IEndpointStateChangeSubscriber, IFai
parentRepairSessions.clear();
}
public void recordRepairStatus(int cmd, ParentRepairStatus parentRepairStatus, List<String> messages)
{
repairStatusByCmd.put(cmd, Pair.create(parentRepairStatus, messages));
}
Pair<ParentRepairStatus, List<String>> getRepairStatus(Integer cmd)
{
return repairStatusByCmd.getIfPresent(cmd);
}
/**
* Return all of the neighbors with whom we share the provided range.
*

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

@ -3349,6 +3349,15 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
ActiveRepairService.instance.terminateSessions();
}
@Nullable
public List<String> getParentRepairStatus(int cmd)
{
Pair<ActiveRepairService.ParentRepairStatus, List<String>> pair = ActiveRepairService.instance.getRepairStatus(cmd);
return pair == null ? null :
ImmutableList.<String>builder().add(pair.left.name()).addAll(pair.right).build();
}
/* End of MBean interface methods */
/**

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

@ -26,7 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nullable;
import javax.management.NotificationEmitter;
import javax.management.openmbean.TabularData;
@ -324,6 +324,15 @@ public interface StorageServiceMBean extends NotificationEmitter
public void forceTerminateAllRepairSessions();
/**
* Get the status of a given parent repair session.
* @param cmd the int reference returned when issuing the repair
* @return status of parent repair from enum {@link org.apache.cassandra.repair.RepairRunnable.Status}
* followed by final message or messages of the session
*/
@Nullable
public List<String> getParentRepairStatus(int cmd);
/**
* transfer this node's data to other machines and remove it from service.
* @param force Decommission even if this will reduce N to be less than RF.

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

@ -103,6 +103,9 @@ public class NodeProbe implements AutoCloseable
private static final String fmtUrl = "service:jmx:rmi:///jndi/rmi://[%s]:%d/jmxrmi";
private static final String ssObjName = "org.apache.cassandra.db:type=StorageService";
private static final int defaultPort = 7199;
static long JMX_NOTIFICATION_POLL_INTERVAL_SECONDS = Long.getLong("cassandra.nodetool.jmx_notification_poll_interval_seconds", TimeUnit.SECONDS.convert(5, TimeUnit.MINUTES));
final String host;
final int port;
private String username;

73
src/java/org/apache/cassandra/tools/RepairRunner.java
View File

@ -20,9 +20,12 @@ package org.apache.cassandra.tools;
import java.io.IOException;
import java.io.PrintStream;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.progress.ProgressEvent;
@ -40,7 +43,6 @@ public class RepairRunner extends JMXNotificationProgressListener
private final Condition condition = new SimpleCondition();
private int cmd;
private volatile boolean hasNotificationLost;
private volatile Exception error;
public RepairRunner(PrintStream out, StorageServiceMBean ssProxy, String keyspace, Map<String, String> options)
@ -62,15 +64,15 @@ public class RepairRunner extends JMXNotificationProgressListener
}
else
{
condition.await();
while (!condition.await(NodeProbe.JMX_NOTIFICATION_POLL_INTERVAL_SECONDS, TimeUnit.SECONDS))
{
queryForCompletedRepair(String.format("After waiting for poll interval of %s seconds",
NodeProbe.JMX_NOTIFICATION_POLL_INTERVAL_SECONDS));
}
if (error != null)
{
throw error;
}
if (hasNotificationLost)
{
out.println(String.format("There were some lost notification(s). You should check server log for repair status of keyspace %s", keyspace));
}
}
}
@ -83,7 +85,11 @@ public class RepairRunner extends JMXNotificationProgressListener
@Override
public void handleNotificationLost(long timestamp, String message)
{
hasNotificationLost = true;
if (cmd > 0)
{
// Check to see if the lost notification was a completion message
queryForCompletedRepair("After receiving lost notification");
}
}
@Override
@ -96,8 +102,8 @@ public class RepairRunner extends JMXNotificationProgressListener
public void handleConnectionFailed(long timestamp, String message)
{
error = new IOException(String.format("[%s] JMX connection closed. You should check server log for repair status of keyspace %s"
+ "(Subsequent keyspaces are not going to be repaired).",
format.format(timestamp), keyspace));
+ "(Subsequent keyspaces are not going to be repaired).",
format.format(timestamp), keyspace));
condition.signalAll();
}
@ -108,7 +114,7 @@ public class RepairRunner extends JMXNotificationProgressListener
String message = String.format("[%s] %s", format.format(System.currentTimeMillis()), event.getMessage());
if (type == ProgressEventType.PROGRESS)
{
message = message + " (progress: " + (int)event.getProgressPercentage() + "%)";
message = message + " (progress: " + (int) event.getProgressPercentage() + "%)";
}
out.println(message);
if (type == ProgressEventType.ERROR)
@ -120,4 +126,51 @@ public class RepairRunner extends JMXNotificationProgressListener
condition.signalAll();
}
}
private void queryForCompletedRepair(String triggeringCondition)
{
List<String> status = ssProxy.getParentRepairStatus(cmd);
String queriedString = "queried for parent session status and";
if (status == null)
{
String message = String.format("[%s] %s %s couldn't find repair status for cmd: %s", triggeringCondition,
queriedString, format.format(System.currentTimeMillis()), cmd);
out.println(message);
}
else
{
ActiveRepairService.ParentRepairStatus parentRepairStatus = ActiveRepairService.ParentRepairStatus.valueOf(status.get(0));
List<String> messages = status.subList(1, status.size());
switch (parentRepairStatus)
{
case COMPLETED:
case FAILED:
out.println(String.format("[%s] %s %s discovered repair %s.",
this.format.format(System.currentTimeMillis()), triggeringCondition,
queriedString, parentRepairStatus.name().toLowerCase()));
if (parentRepairStatus == ActiveRepairService.ParentRepairStatus.FAILED)
{
error = new IOException(messages.get(0));
}
printMessages(messages);
condition.signalAll();
break;
case IN_PROGRESS:
break;
default:
out.println(String.format("[%s] WARNING Encountered unexpected RepairRunnable.ParentRepairStatus: %s", System.currentTimeMillis(), parentRepairStatus));
printMessages(messages);
break;
}
}
}
private void printMessages(List<String> messages)
{
for (String message : messages)
{
out.println(String.format("[%s] %s", this.format.format(System.currentTimeMillis()), message));
}
}
}

26
test/unit/org/apache/cassandra/service/ActiveRepairServiceTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.service;
import java.net.InetAddress;
import java.util.*;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Sets;
import org.junit.Before;
import org.junit.BeforeClass;
@ -45,8 +46,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Refs;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
public class ActiveRepairServiceTest
{
@ -211,6 +212,27 @@ public class ActiveRepairServiceTest
ActiveRepairService.getNeighbors(KEYSPACE5, ranges, ranges.iterator().next(), null, hosts);
}
@Test
public void testParentRepairStatus() throws Throwable
{
ActiveRepairService.instance.recordRepairStatus(1, ActiveRepairService.ParentRepairStatus.COMPLETED, ImmutableList.of("foo", "bar"));
List<String> res = StorageService.instance.getParentRepairStatus(1);
assertNotNull(res);
assertEquals(ActiveRepairService.ParentRepairStatus.COMPLETED, ActiveRepairService.ParentRepairStatus.valueOf(res.get(0)));
assertEquals("foo", res.get(1));
assertEquals("bar", res.get(2));
List<String> emptyRes = StorageService.instance.getParentRepairStatus(44);
assertNull(emptyRes);
ActiveRepairService.instance.recordRepairStatus(3, ActiveRepairService.ParentRepairStatus.FAILED, ImmutableList.of("some failure message", "bar"));
List<String> failed = StorageService.instance.getParentRepairStatus(3);
assertNotNull(failed);
assertEquals(ActiveRepairService.ParentRepairStatus.FAILED, ActiveRepairService.ParentRepairStatus.valueOf(failed.get(0)));
}
Set<InetAddress> addTokens(int max) throws Throwable
{
TokenMetadata tmd = StorageService.instance.getTokenMetadata();