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=9faffb9395dcdb98d02951e7e9530d1ac602b3f5;hp=3dc6ae469a932474effca186c956a7a6f4ec8631;hb=79c3fb9269d9baeb3a2787544fca3636e0ea608f;hpb=c64ef5f44f131976c20fcf8ced56627f81091838 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 3dc6ae469a..9faffb9395 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 @@ -19,28 +19,36 @@ import akka.persistence.SnapshotSelectionCriteria; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Optional; import com.google.common.base.Stopwatch; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Lists; import com.google.protobuf.ByteString; import java.io.Serializable; +import java.util.Collection; +import java.util.List; import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.time.DurationFormatUtils; 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; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; import org.opendaylight.controller.cluster.raft.base.messages.Replicate; -import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat; import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot; +import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader; import org.opendaylight.controller.cluster.raft.behaviors.AbstractRaftActorBehavior; import org.opendaylight.controller.cluster.raft.behaviors.Follower; import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; import org.opendaylight.controller.cluster.raft.client.messages.FindLeader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply; -import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; +import org.opendaylight.controller.cluster.raft.client.messages.FollowerInfo; +import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState; +import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; -import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -85,6 +93,16 @@ 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 + + private static final Procedure APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK = + new Procedure() { + @Override + public void apply(ApplyJournalEntries param) throws Exception { + } + }; + protected final Logger LOG = LoggerFactory.getLogger(getClass()); /** @@ -97,7 +115,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 @@ -126,8 +144,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { private void initRecoveryTimer() { if(recoveryTimer == null) { - recoveryTimer = new Stopwatch(); - recoveryTimer.start(); + recoveryTimer = Stopwatch.createStarted(); } } @@ -139,6 +156,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()) { @@ -147,7 +177,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) { @@ -190,8 +223,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { context.setLastApplied(snapshot.getLastAppliedIndex()); context.setCommitIndex(snapshot.getLastAppliedIndex()); - Stopwatch timer = new Stopwatch(); - timer.start(); + Stopwatch timer = Stopwatch.createStarted(); // Apply the snapshot to the actors state applyRecoverySnapshot(snapshot.getState()); @@ -210,18 +242,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) { @@ -283,6 +315,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(), @@ -292,16 +330,13 @@ 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() { - @Override - public void apply(ApplyLogEntries param) throws Exception { - } - }); + + persistence().persist(applyEntries, APPLY_JOURNAL_ENTRIES_PERSIST_CALLBACK); } else if(message instanceof ApplySnapshot ) { Snapshot snapshot = ((ApplySnapshot) message).getSnapshot(); @@ -349,7 +384,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { context.getReplicatedLog().size()); } else if (message instanceof CaptureSnapshot) { - LOG.info("{}: CaptureSnapshot received by actor", persistenceId()); + LOG.debug("{}: CaptureSnapshot received by actor: {}", persistenceId(), message); if(captureSnapshot == null) { captureSnapshot = (CaptureSnapshot)message; @@ -358,15 +393,9 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } else if (message instanceof CaptureSnapshotReply){ handleCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot()); - + } else if(message instanceof GetOnDemandRaftState) { + onGetOnDemandRaftStats(); } else { - if (!(message instanceof AppendEntriesMessages.AppendEntries) - && !(message instanceof AppendEntriesReply) && !(message instanceof SendHeartBeat)) { - if(LOG.isDebugEnabled()) { - LOG.debug("{}: onReceiveCommand: message: {}", persistenceId(), message.getClass()); - } - } - RaftActorBehavior oldBehavior = currentBehavior; currentBehavior = currentBehavior.handleMessage(getSender(), message); @@ -374,6 +403,49 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } } + private void onGetOnDemandRaftStats() { + // Debugging message to retrieve raft stats. + + OnDemandRaftState.Builder builder = OnDemandRaftState.builder() + .commitIndex(context.getCommitIndex()) + .currentTerm(context.getTermInformation().getCurrentTerm()) + .inMemoryJournalDataSize(replicatedLog.dataSize()) + .inMemoryJournalLogSize(replicatedLog.size()) + .isSnapshotCaptureInitiated(context.isSnapshotCaptureInitiated()) + .lastApplied(context.getLastApplied()) + .lastIndex(replicatedLog.lastIndex()) + .lastTerm(replicatedLog.lastTerm()) + .leader(getLeaderId()) + .raftState(currentBehavior.state().toString()) + .replicatedToAllIndex(currentBehavior.getReplicatedToAllIndex()) + .snapshotIndex(replicatedLog.getSnapshotIndex()) + .snapshotTerm(replicatedLog.getSnapshotTerm()) + .votedFor(context.getTermInformation().getVotedFor()) + .peerAddresses(ImmutableMap.copyOf(context.getPeerAddresses())); + + ReplicatedLogEntry lastLogEntry = getLastLogEntry(); + if (lastLogEntry != null) { + builder.lastLogIndex(lastLogEntry.getIndex()); + builder.lastLogTerm(lastLogEntry.getTerm()); + } + + if(currentBehavior instanceof AbstractLeader) { + AbstractLeader leader = (AbstractLeader)currentBehavior; + Collection followerIds = leader.getFollowerIds(); + List followerInfoList = Lists.newArrayListWithCapacity(followerIds.size()); + for(String id: followerIds) { + final FollowerLogInformation info = leader.getFollower(id); + followerInfoList.add(new FollowerInfo(id, info.getNextIndex(), info.getMatchIndex(), + info.isFollowerActive(), DurationFormatUtils.formatDurationHMS(info.timeSinceLastActivity()))); + } + + builder.followerInfoList(followerInfoList); + } + + sender().tell(builder.build(), self()); + + } + private void handleBehaviorChange(RaftActorBehavior oldBehavior, RaftActorBehavior currentBehavior) { if (oldBehavior != currentBehavior){ onStateChanged(); @@ -426,9 +498,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()){ @@ -513,6 +585,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. *

@@ -575,7 +651,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { /** * This method is called during recovery to reconstruct the state of the actor. * - * @param snapshot A snapshot of the state of the actor + * @param snapshotBytes A snapshot of the state of the actor */ protected abstract void applyRecoverySnapshot(byte[] snapshotBytes); @@ -656,7 +732,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } private void handleCaptureSnapshotReply(byte[] snapshotBytes) { - LOG.info("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length); + LOG.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", persistenceId(), snapshotBytes.length); // create a snapshot object from the state provided and save it // when snapshot is saved async, SaveSnapshotSuccess is raised. @@ -670,16 +746,46 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { LOG.info("{}: Persisting of snapshot done:{}", persistenceId(), sn.getLogMessage()); - //be greedy and remove entries from in-mem journal which are in the snapshot - // and update snapshotIndex and snapshotTerm without waiting for the success, + long dataThreshold = getTotalMemory() * + getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100; + if (context.getReplicatedLog().dataSize() > dataThreshold) { + + if(LOG.isDebugEnabled()) { + LOG.debug("{}: dataSize {} exceeds dataThreshold {} - doing snapshotPreCommit with index {}", + persistenceId(), context.getReplicatedLog().dataSize(), dataThreshold, + captureSnapshot.getLastAppliedIndex()); + } - context.getReplicatedLog().snapshotPreCommit( - captureSnapshot.getLastAppliedIndex(), - captureSnapshot.getLastAppliedTerm()); + // if memory is less, clear the log based on lastApplied. + // this could/should only happen if one of the followers is down + // as normally we keep removing from the log when its replicated to all. + context.getReplicatedLog().snapshotPreCommit(captureSnapshot.getLastAppliedIndex(), + captureSnapshot.getLastAppliedTerm()); - LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex:{} " + - "and term:{}", persistenceId(), captureSnapshot.getLastAppliedIndex(), - captureSnapshot.getLastAppliedTerm()); + // Don't reset replicatedToAllIndex to -1 as this may prevent us from trimming the log after an + // install snapshot to a follower. + if(captureSnapshot.getReplicatedToAllIndex() >= 0) { + 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()); + } + + + LOG.info("{}: Removed in-memory snapshotted entries, adjusted snaphsotIndex: {} " + + "and term: {}", persistenceId(), replicatedLog.getSnapshotIndex(), + replicatedLog.getSnapshotTerm()); if (isLeader() && captureSnapshot.isInstallSnapshotInitiated()) { // this would be call straight to the leader and won't initiate in serialization @@ -691,6 +797,10 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { context.setSnapshotCaptureInitiated(false); } + protected long getTotalMemory() { + return Runtime.getRuntime().totalMemory(); + } + protected boolean hasFollowers(){ return getRaftActorContext().getPeerAddresses().keySet().size() > 0; } @@ -719,13 +829,14 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { // FIXME: Maybe this should be done after the command is saved journal.subList(adjustedIndex , journal.size()).clear(); - persistence().persist(new DeleteEntries(adjustedIndex), new Procedure(){ + persistence().persist(new DeleteEntries(adjustedIndex), new Procedure() { - @Override public void apply(DeleteEntries param) - throws Exception { + @Override + public void apply(DeleteEntries param) + throws Exception { //FIXME : Doing nothing for now dataSize = 0; - for(ReplicatedLogEntry entry : journal){ + for (ReplicatedLogEntry entry : journal) { dataSize += entry.size(); } } @@ -737,11 +848,6 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { appendAndPersist(replicatedLogEntry, null); } - @Override - public int dataSize() { - return dataSize; - } - public void appendAndPersist( final ReplicatedLogEntry replicatedLogEntry, final Procedure callback) { @@ -768,7 +874,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { long dataSizeForCheck = dataSize; dataSizeSinceLastSnapshot += logEntrySize; - long journalSize = lastIndex()+1; + long journalSize = lastIndex() + 1; if(!hasFollowers()) { // When we do not have followers we do not maintain an in-memory log @@ -785,7 +891,7 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { dataSizeForCheck = dataSizeSinceLastSnapshot / DATA_SIZE_DIVIDER; } - long dataThreshold = Runtime.getRuntime().totalMemory() * + long dataThreshold = getTotalMemory() * getRaftActorContext().getConfigParams().getSnapshotDataThresholdPercentage() / 100; // when a snaphsot is being taken, captureSnapshot != null @@ -795,7 +901,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; @@ -819,12 +927,15 @@ public abstract class RaftActor extends AbstractUntypedPersistentActor { } // send a CaptureSnapshot to self to make the expensive operation async. - getSelf().tell(new CaptureSnapshot( - lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm), + long replicatedToAllIndex = getCurrentBehavior().getReplicatedToAllIndex(); + ReplicatedLogEntry replicatedToAllEntry = context.getReplicatedLog().get(replicatedToAllIndex); + getSelf().tell(new CaptureSnapshot(lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm, + (replicatedToAllEntry != null ? replicatedToAllEntry.getIndex() : -1), + (replicatedToAllEntry != null ? replicatedToAllEntry.getTerm() : -1)), null); context.setSnapshotCaptureInitiated(true); } - if(callback != null){ + if (callback != null){ callback.apply(replicatedLogEntry); } }