import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
import com.google.protobuf.ByteString;
import java.io.IOException;
import java.util.Collection;
import org.opendaylight.controller.cluster.raft.ClientRequestTrackerImpl;
import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
+import org.opendaylight.controller.cluster.raft.PeerInfo;
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.Snapshot;
+import org.opendaylight.controller.cluster.raft.VotingState;
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.messages.InstallSnapshotReply;
import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
+import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply;
import scala.concurrent.duration.FiniteDuration;
/**
// This would be passed as the hash code of the last chunk when sending the first chunk
public static final int INITIAL_LAST_CHUNK_HASH_CODE = -1;
- private final Map<String, FollowerLogInformation> followerToLog;
+ private final Map<String, FollowerLogInformation> followerToLog = new HashMap<>();
private final Map<String, FollowerToSnapshot> mapFollowerToSnapshot = new HashMap<>();
private Cancellable heartbeatSchedule = null;
private final Collection<ClientRequestTracker> trackerList = new LinkedList<>();
- protected final int minReplicationCount;
-
- protected final int minIsolatedLeaderPeerCount;
+ private int minReplicationCount;
private Optional<SnapshotHolder> snapshot;
setLeaderPayloadVersion(context.getPayloadVersion());
- final Builder<String, FollowerLogInformation> ftlBuilder = ImmutableMap.builder();
- for (String followerId : context.getPeerAddresses().keySet()) {
- FollowerLogInformation followerLogInformation =
- new FollowerLogInformationImpl(followerId, -1, context);
-
- ftlBuilder.put(followerId, followerLogInformation);
+ for(PeerInfo peerInfo: context.getPeers()) {
+ FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(peerInfo, -1, context);
+ followerToLog.put(peerInfo.getId(), followerLogInformation);
}
- followerToLog = ftlBuilder.build();
leaderId = context.getId();
LOG.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds());
- minReplicationCount = getMajorityVoteCount(getFollowerIds().size());
-
- // the isolated Leader peer count will be 1 less than the majority vote count.
- // this is because the vote count has the self vote counted in it
- // for e.g
- // 0 peers = 1 votesRequired , minIsolatedLeaderPeerCount = 0
- // 2 peers = 2 votesRequired , minIsolatedLeaderPeerCount = 1
- // 4 peers = 3 votesRequired, minIsolatedLeaderPeerCount = 2
- minIsolatedLeaderPeerCount = minReplicationCount > 0 ? (minReplicationCount - 1) : 0;
+ updateMinReplicaCount();
snapshot = Optional.absent();
return followerToLog.keySet();
}
+ public void addFollower(String followerId) {
+ FollowerLogInformation followerLogInformation = new FollowerLogInformationImpl(
+ context.getPeerInfo(followerId), -1, context);
+ followerToLog.put(followerId, followerLogInformation);
+
+ if(heartbeatSchedule == null) {
+ scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval());
+ }
+ }
+
+ public void removeFollower(String followerId) {
+ followerToLog.remove(followerId);
+ }
+
+ public void updateMinReplicaCount() {
+ int numVoting = 0;
+ for(PeerInfo peer: context.getPeers()) {
+ if(peer.isVoting()) {
+ numVoting++;
+ }
+ }
+
+ minReplicationCount = getMajorityVoteCount(numVoting);
+ }
+
+ protected int getMinIsolatedLeaderPeerCount(){
+ //the isolated Leader peer count will be 1 less than the majority vote count.
+ //this is because the vote count has the self vote counted in it
+ //for e.g
+ //0 peers = 1 votesRequired , minIsolatedLeaderPeerCount = 0
+ //2 peers = 2 votesRequired , minIsolatedLeaderPeerCount = 1
+ //4 peers = 3 votesRequired, minIsolatedLeaderPeerCount = 2
+
+ return minReplicationCount > 0 ? (minReplicationCount - 1) : 0;
+ }
+
@VisibleForTesting
void setSnapshot(@Nullable Snapshot snapshot) {
if(snapshot != null) {
long followerLastLogIndex = appendEntriesReply.getLogLastIndex();
ReplicatedLogEntry followersLastLogEntry = context.getReplicatedLog().get(followerLastLogIndex);
- if(followerLastLogIndex < 0 || (followersLastLogEntry != null &&
+ if(appendEntriesReply.isForceInstallSnapshot()) {
+ // Reset the followers match and next index. This is to signal that this follower has nothing
+ // in common with this Leader and so would require a snapshot to be installed
+ followerLogInformation.setMatchIndex(-1);
+ followerLogInformation.setNextIndex(-1);
+
+ // Force initiate a snapshot capture
+ initiateCaptureSnapshot(followerId);
+ } else if(followerLastLogIndex < 0 || (followersLastLogEntry != null &&
followersLastLogEntry.getTerm() == appendEntriesReply.getLogLastTerm())) {
// The follower's log is empty or the last entry is present in the leader's journal
// and the terms match so the follower is just behind the leader's journal from
if (replicatedCount >= minReplicationCount) {
ReplicatedLogEntry replicatedLogEntry = context.getReplicatedLog().get(N);
- if (replicatedLogEntry != null &&
- replicatedLogEntry.getTerm() == currentTerm()) {
+ if (replicatedLogEntry != null && replicatedLogEntry.getTerm() == currentTerm()) {
context.setCommitIndex(N);
+ } else {
+ break;
}
} else {
break;
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
- return switchBehavior(new Follower(context));
+ return internalSwitchBehavior(RaftState.Follower);
}
}
mapFollowerToSnapshot.remove(followerId);
LOG.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}",
- logName(), followerId, followerLogInformation.getMatchIndex(),
- followerLogInformation.getNextIndex());
+ logName(), followerId, followerLogInformation.getMatchIndex(),
+ followerLogInformation.getNextIndex());
if (mapFollowerToSnapshot.isEmpty()) {
// once there are no pending followers receiving snapshots
setSnapshot(null);
}
wasLastChunk = true;
-
+ if(context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED){
+ UnInitializedFollowerSnapshotReply unInitFollowerSnapshotSuccess =
+ new UnInitializedFollowerSnapshotReply(followerId);
+ context.getActor().tell(unInitFollowerSnapshotSuccess, context.getActor());
+ LOG.debug("Sent message UnInitializedFollowerSnapshotReply to self");
+ }
} else {
followerToSnapshot.markSendStatus(true);
}
logIndex)
);
- if (followerToLog.isEmpty()) {
+ boolean applyModificationToState = followerToLog.isEmpty()
+ || context.getRaftPolicy().applyModificationToStateBeforeConsensus();
+
+ if(applyModificationToState){
context.setCommitIndex(logIndex);
applyLogToStateMachine(logIndex);
- } else {
+ }
+
+ if (!followerToLog.isEmpty()) {
sendAppendEntries(0, false);
}
}
// Send heartbeat to follower whenever install snapshot is initiated.
sendAppendEntries = true;
- initiateCaptureSnapshot(followerId, followerNextIndex);
+ if (canInstallSnapshot(followerNextIndex)) {
+ initiateCaptureSnapshot(followerId);
+ }
} else if(sendHeartbeat) {
// we send an AppendEntries, even if the follower is inactive
* 6. If another follower requires a snapshot and a snapshot has been collected (via CaptureSnapshotReply)
* then send the existing snapshot in chunks to the follower.
* @param followerId
- * @param followerNextIndex
*/
- private void initiateCaptureSnapshot(String followerId, long followerNextIndex) {
- if (!context.getReplicatedLog().isPresent(followerNextIndex) &&
- context.getReplicatedLog().isInSnapshot(followerNextIndex)) {
+ public void initiateCaptureSnapshot(String followerId) {
+ if (snapshot.isPresent()) {
+ // if a snapshot is present in the memory, most likely another install is in progress
+ // no need to capture snapshot.
+ // This could happen if another follower needs an install when one is going on.
+ final ActorSelection followerActor = context.getPeerActorSelection(followerId);
+ sendSnapshotChunk(followerActor, followerId);
- if (snapshot.isPresent()) {
- // if a snapshot is present in the memory, most likely another install is in progress
- // no need to capture snapshot.
- // This could happen if another follower needs an install when one is going on.
- final ActorSelection followerActor = context.getPeerActorSelection(followerId);
- sendSnapshotChunk(followerActor, followerId);
-
- } else {
- context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(),
- this.getReplicatedToAllIndex(), followerId);
- }
+ } else {
+ context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(),
+ this.getReplicatedToAllIndex(), followerId);
}
}
+ private boolean canInstallSnapshot(long nextIndex){
+ // If the follower's nextIndex is -1 then we might as well send it a snapshot
+ // Otherwise send it a snapshot only if the nextIndex is not present in the log but is present
+ // in the snapshot
+ return (nextIndex == -1 ||
+ (!context.getReplicatedLog().isPresent(nextIndex)
+ && context.getReplicatedLog().isInSnapshot(nextIndex)));
+
+ }
+
private void sendInstallSnapshot() {
LOG.debug("{}: sendInstallSnapshot", logName());
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
- ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
+ String followerId = e.getKey();
+ ActorSelection followerActor = context.getPeerActorSelection(followerId);
+ FollowerLogInformation followerLogInfo = e.getValue();
if (followerActor != null) {
- long nextIndex = e.getValue().getNextIndex();
-
- if (!context.getReplicatedLog().isPresent(nextIndex) &&
- context.getReplicatedLog().isInSnapshot(nextIndex)) {
- sendSnapshotChunk(followerActor, e.getKey());
+ long nextIndex = followerLogInfo.getNextIndex();
+ if (context.getPeerInfo(followerId).getVotingState() == VotingState.VOTING_NOT_INITIALIZED ||
+ canInstallSnapshot(nextIndex)) {
+ sendSnapshotChunk(followerActor, followerId);
}
}
}
}
protected boolean isLeaderIsolated() {
- int minPresent = minIsolatedLeaderPeerCount;
+ int minPresent = getMinIsolatedLeaderPeerCount();
for (FollowerLogInformation followerLogInformation : followerToLog.values()) {
if (followerLogInformation.isFollowerActive()) {
--minPresent;