X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FFollower.java;h=82c6bafedd19b7a4337f802539ade9211b0280a0;hb=01c5a7cc52f8a438053d52ebb91e531493ca765a;hp=974ec475856af94f6e36ad367b7cf555d71354bd;hpb=a93bcbe711f66ef6ec7bc97972f108859c87a11e;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java index 974ec47585..82c6bafedd 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Follower.java @@ -9,15 +9,16 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; +import akka.japi.Procedure; import com.google.common.annotations.VisibleForTesting; import java.util.ArrayList; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload; import org.opendaylight.controller.cluster.raft.Snapshot; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; -import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot; @@ -37,18 +38,31 @@ import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; * */ public class Follower extends AbstractRaftActorBehavior { + private static final int SYNC_THRESHOLD = 10; - private SnapshotTracker snapshotTracker = null; + private final SyncStatusTracker initialSyncStatusTracker; - private final InitialSyncStatusTracker initialSyncStatusTracker; + private final Procedure appendAndPersistCallback = new Procedure() { + @Override + public void apply(ReplicatedLogEntry logEntry) { + context.getReplicatedLog().captureSnapshotIfReady(logEntry); + } + }; + + private SnapshotTracker snapshotTracker = null; public Follower(RaftActorContext context) { + this(context, null); + } + + public Follower(RaftActorContext context, String initialLeaderId) { super(context, RaftState.Follower); + leaderId = initialLeaderId; - initialSyncStatusTracker = new InitialSyncStatusTracker(context.getActor()); + initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), SYNC_THRESHOLD); - if(context.getRaftPolicy().automaticElectionsEnabled()) { - if (context.getPeerAddresses().isEmpty()) { + if(canStartElection()) { + if (context.getPeerIds().isEmpty() && getLeaderId() == null) { actor().tell(ELECTION_TIMEOUT, actor()); } else { scheduleElection(electionDuration()); @@ -102,7 +116,7 @@ public class Follower extends AbstractRaftActorBehavior { // to make it easier to read. Before refactoring ensure tests // cover the code properly - if (snapshotTracker != null) { + if (snapshotTracker != null || context.getSnapshotManager().isApplying()) { // if snapshot install is in progress, follower should just acknowledge append entries with a reply. AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), true, lastIndex(), lastTerm(), context.getPayloadVersion()); @@ -187,7 +201,11 @@ public class Follower extends AbstractRaftActorBehavior { LOG.debug("{}: Append entry to log {}", logName(), entry.getData()); - context.getReplicatedLog().appendAndPersist(entry); + context.getReplicatedLog().appendAndPersist(entry, appendAndPersistCallback); + + if(entry.getData() instanceof ServerConfigurationPayload) { + context.updatePeerIds((ServerConfigurationPayload)entry.getData()); + } } LOG.debug("{}: Log size is now {}", logName(), context.getReplicatedLog().size()); @@ -259,8 +277,8 @@ public class Follower extends AbstractRaftActorBehavior { // The follower's log is out of sync because the Leader's // prevLogIndex entry was not found in it's log - LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it", - logName(), appendEntries.getPrevLogIndex()); + LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it - lastIndex: {}, snapshotIndex: {}", + logName(), appendEntries.getPrevLogIndex(), lastIndex, context.getReplicatedLog().getSnapshotIndex()); } else if (lastIndex > -1 && prevEntryPresent && prevLogTerm != appendEntries.getPrevLogTerm()) { // The follower's log is out of sync because the Leader's @@ -272,7 +290,8 @@ public class Follower extends AbstractRaftActorBehavior { logName(), prevLogTerm, appendEntries.getPrevLogTerm()); } else if(appendEntries.getPrevLogIndex() == -1 && appendEntries.getPrevLogTerm() == -1 && appendEntries.getReplicatedToAllIndex() != -1 - && !isLogEntryPresent(appendEntries.getReplicatedToAllIndex())) { + && !isLogEntryPresent(appendEntries.getReplicatedToAllIndex()) + && !context.getReplicatedLog().isInSnapshot(appendEntries.getReplicatedToAllIndex())) { // This append entry comes from a leader who has it's log aggressively trimmed and so does not have // the previous entry in it's in-memory journal @@ -280,8 +299,9 @@ public class Follower extends AbstractRaftActorBehavior { "{}: Cannot append entries because the replicatedToAllIndex {} does not appear to be in the in-memory journal", logName(), appendEntries.getReplicatedToAllIndex()); } else if(appendEntries.getPrevLogIndex() == -1 && appendEntries.getPrevLogTerm() == -1 - && appendEntries.getReplicatedToAllIndex() != -1 && numLogEntries > 0 && - !isLogEntryPresent(appendEntries.getEntries().get(0).getIndex() - 1)){ + && appendEntries.getReplicatedToAllIndex() != -1 && numLogEntries > 0 + && !isLogEntryPresent(appendEntries.getEntries().get(0).getIndex() - 1) + && !context.getReplicatedLog().isInSnapshot(appendEntries.getEntries().get(0).getIndex() - 1)) { LOG.debug( "{}: Cannot append entries because the calculated previousIndex {} was not found in the in-memory journal", logName(), appendEntries.getEntries().get(0).getIndex() - 1); @@ -319,8 +339,12 @@ public class Follower extends AbstractRaftActorBehavior { } if (message instanceof ElectionTimeout) { - LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName()); - return internalSwitchBehavior(RaftState.Candidate); + if(canStartElection()) { + LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName()); + return internalSwitchBehavior(RaftState.Candidate); + } else { + return this; + } } else if (message instanceof InstallSnapshot) { InstallSnapshot installSnapshot = (InstallSnapshot) message; @@ -334,12 +358,14 @@ public class Follower extends AbstractRaftActorBehavior { return super.handleMessage(sender, message); } - private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) { + private void handleInstallSnapshot(final ActorRef sender, InstallSnapshot installSnapshot) { LOG.debug("{}: InstallSnapshot received from leader {}, datasize: {} , Chunk: {}/{}", logName(), installSnapshot.getLeaderId(), installSnapshot.getData().size(), installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks()); + leaderId = installSnapshot.getLeaderId(); + if(snapshotTracker == null){ snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks()); } @@ -347,6 +373,9 @@ public class Follower extends AbstractRaftActorBehavior { updateInitialSyncStatus(installSnapshot.getLastIncludedIndex(), installSnapshot.getLeaderId()); try { + final InstallSnapshotReply reply = new InstallSnapshotReply( + currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true); + if(snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(), installSnapshot.getLastChunkHashCode())){ Snapshot snapshot = Snapshot.create(snapshotTracker.getSnapshot(), @@ -354,21 +383,33 @@ public class Follower extends AbstractRaftActorBehavior { installSnapshot.getLastIncludedIndex(), installSnapshot.getLastIncludedTerm(), installSnapshot.getLastIncludedIndex(), - installSnapshot.getLastIncludedTerm()); - - actor().tell(new ApplySnapshot(snapshot), actor()); + installSnapshot.getLastIncludedTerm(), + context.getTermInformation().getCurrentTerm(), + context.getTermInformation().getVotedFor(), + context.getPeerServerInfo(true)); - snapshotTracker = null; + ApplySnapshot.Callback applySnapshotCallback = new ApplySnapshot.Callback() { + @Override + public void onSuccess() { + LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply); - } + sender.tell(reply, actor()); + } - InstallSnapshotReply reply = new InstallSnapshotReply( - currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true); + @Override + public void onFailure() { + sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), -1, false), actor()); + } + }; - LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply); + actor().tell(new ApplySnapshot(snapshot, applySnapshotCallback), actor()); - sender.tell(reply, actor()); + snapshotTracker = null; + } else { + LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply); + sender.tell(reply, actor()); + } } catch (SnapshotTracker.InvalidChunkException e) { LOG.debug("{}: Exception in InstallSnapshot of follower", logName(), e); @@ -395,36 +436,4 @@ public class Follower extends AbstractRaftActorBehavior { SnapshotTracker getSnapshotTracker(){ return snapshotTracker; } - - private class InitialSyncStatusTracker { - - private static final long INVALID_LOG_INDEX = -2L; - private long initialLeaderCommit = INVALID_LOG_INDEX; - private boolean initialSyncUpDone = false; - private String syncedLeaderId = null; - private final ActorRef actor; - - public InitialSyncStatusTracker(ActorRef actor) { - this.actor = actor; - } - - public void update(String leaderId, long leaderCommit, long commitIndex){ - - if(!leaderId.equals(syncedLeaderId)){ - initialSyncUpDone = false; - initialLeaderCommit = INVALID_LOG_INDEX; - syncedLeaderId = leaderId; - } - - if(!initialSyncUpDone){ - if(initialLeaderCommit == INVALID_LOG_INDEX){ - actor.tell(new FollowerInitialSyncUpStatus(false, getId()), ActorRef.noSender()); - initialLeaderCommit = leaderCommit; - } else if(commitIndex >= initialLeaderCommit){ - actor.tell(new FollowerInitialSyncUpStatus(true, getId()), ActorRef.noSender()); - initialSyncUpDone = true; - } - } - } - } }