X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActor.java;h=ec3f375bdeb0fb940e4d08d14397ebd03f957853;hp=854ceb23d047fabea219f3861c5f44c0f2afc907;hb=f78b7f15e24efdb5dd9f91b487bc63dad7517b1c;hpb=614324d63a339ef4acbc9e2c3bbaaef469f97868 diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java index 854ceb23d0..ec3f375bde 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActor.java @@ -22,9 +22,11 @@ import com.google.common.base.Stopwatch; import com.google.protobuf.ByteString; import java.io.Serializable; import java.util.Map; +import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActor; import org.opendaylight.controller.cluster.notifications.RoleChanged; +import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; @@ -82,6 +84,9 @@ import org.slf4j.LoggerFactory; * */ public abstract class RaftActor extends AbstractUntypedPersistentActor { + + private static final long APPLY_STATE_DELAY_THRESHOLD_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(50L); // 50 millis + protected final Logger LOG = LoggerFactory.getLogger(getClass()); /** @@ -94,7 +99,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { * This context should NOT be passed directly to any other actor it is * only to be consumed by the RaftActorBehaviors */ - private final RaftActorContext context; + private final RaftActorContextImpl context; /** * The in-memory journal @@ -135,6 +140,19 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { super.preStart(); } + @Override + public void postStop() { + if(currentBehavior != null) { + try { + currentBehavior.close(); + } catch (Exception e) { + LOG.debug("{}: Error closing behavior {}", persistenceId(), currentBehavior.state()); + } + } + + super.postStop(); + } + @Override public void handleRecover(Object message) { if(persistence().isRecoveryApplicable()) { @@ -143,7 +161,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } else if (message instanceof ReplicatedLogEntry) { onRecoveredJournalLogEntry((ReplicatedLogEntry) message); } else if (message instanceof ApplyLogEntries) { - onRecoveredApplyLogEntries((ApplyLogEntries) message); + // Handle this message for backwards compatibility with pre-Lithium versions. + onRecoveredApplyLogEntries(((ApplyLogEntries) message).getToIndex()); + } else if (message instanceof ApplyJournalEntries) { + onRecoveredApplyLogEntries(((ApplyJournalEntries) message).getToIndex()); } else if (message instanceof DeleteEntries) { replicatedLog.removeFrom(((DeleteEntries) message).getFromIndex()); } else if (message instanceof UpdateElectionTerm) { @@ -205,18 +226,18 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { replicatedLog.append(logEntry); } - private void onRecoveredApplyLogEntries(ApplyLogEntries ale) { + private void onRecoveredApplyLogEntries(long toIndex) { if(LOG.isDebugEnabled()) { LOG.debug("{}: Received ApplyLogEntries for recovery, applying to state: {} to {}", - persistenceId(), context.getLastApplied() + 1, ale.getToIndex()); + persistenceId(), context.getLastApplied() + 1, toIndex); } - for (long i = context.getLastApplied() + 1; i <= ale.getToIndex(); i++) { + for (long i = context.getLastApplied() + 1; i <= toIndex; i++) { batchRecoveredLogEntry(replicatedLog.get(i)); } - context.setLastApplied(ale.getToIndex()); - context.setCommitIndex(ale.getToIndex()); + context.setLastApplied(toIndex); + context.setCommitIndex(toIndex); } private void batchRecoveredLogEntry(ReplicatedLogEntry logEntry) { @@ -278,6 +299,12 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { if (message instanceof ApplyState){ ApplyState applyState = (ApplyState) message; + long elapsedTime = (System.nanoTime() - applyState.getStartTime()); + if(elapsedTime >= APPLY_STATE_DELAY_THRESHOLD_IN_NANOS){ + LOG.warn("ApplyState took more time than expected. Elapsed Time = {} ms ApplyState = {}", + TimeUnit.NANOSECONDS.toMillis(elapsedTime), applyState); + } + if(LOG.isDebugEnabled()) { LOG.debug("{}: Applying state for log index {} data {}", persistenceId(), applyState.getReplicatedLogEntry().getIndex(), @@ -287,14 +314,14 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { applyState(applyState.getClientActor(), applyState.getIdentifier(), applyState.getReplicatedLogEntry().getData()); - } else if (message instanceof ApplyLogEntries){ - ApplyLogEntries ale = (ApplyLogEntries) message; + } else if (message instanceof ApplyJournalEntries){ + ApplyJournalEntries applyEntries = (ApplyJournalEntries) message; if(LOG.isDebugEnabled()) { - LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), ale.getToIndex()); + LOG.debug("{}: Persisting ApplyLogEntries with index={}", persistenceId(), applyEntries.getToIndex()); } - persistence().persist(new ApplyLogEntries(ale.getToIndex()), new Procedure() { + persistence().persist(applyEntries, new Procedure() { @Override - public void apply(ApplyLogEntries param) throws Exception { + public void apply(ApplyJournalEntries param) throws Exception { } }); @@ -414,9 +441,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { // Apply the state immediately applyState(clientActor, identifier, data); - // Send a ApplyLogEntries message so that we write the fact that we applied + // Send a ApplyJournalEntries message so that we write the fact that we applied // the state to durable storage - self().tell(new ApplyLogEntries((int) replicatedLogEntry.getIndex()), self()); + self().tell(new ApplyJournalEntries(replicatedLogEntry.getIndex()), self()); // Check if the "real" snapshot capture has been initiated. If no then do the fake snapshot if(!context.isSnapshotCaptureInitiated()){ @@ -501,6 +528,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { return context; } + protected void updateConfigParams(ConfigParams configParams) { + context.setConfigParams(configParams); + } + /** * setPeerAddress sets the address of a known peer at a later time. *

@@ -667,12 +698,22 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(), captureSnapshot.getLastAppliedTerm()); - } else { + getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex()); + } else if(captureSnapshot.getReplicatedToAllIndex() != -1){ // clear the log based on replicatedToAllIndex context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getReplicatedToAllIndex(), captureSnapshot.getReplicatedToAllTerm()); + + getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex()); + } else { + // The replicatedToAllIndex was not found in the log + // This means that replicatedToAllIndex never moved beyond -1 or that it is already in the snapshot. + // In this scenario we may need to save the snapshot to the akka persistence + // snapshot for recovery but we do not need to do the replicated log trimming. + context.getReplicatedLog().snapshotPreCommit(replicatedLog.getSnapshotIndex(), + replicatedLog.getSnapshotTerm()); } - getCurrentBehavior().setReplicatedToAllIndex(captureSnapshot.getReplicatedToAllIndex()); + LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " + "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(), @@ -788,7 +829,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { dataSizeSinceLastSnapshot = 0; - LOG.info("{}: Initiating Snapshot Capture..", persistenceId()); + LOG.info("{}: Initiating Snapshot Capture, journalSize = {}, dataSizeForCheck = {}," + + " dataThreshold = {}", persistenceId(), journalSize, dataSizeForCheck, dataThreshold); + long lastAppliedIndex = -1; long lastAppliedTerm = -1;