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%2FAbstractLeader.java;h=a4753a4fe432654554a64578af0b09233069146b;hb=d8d8f731bbe6c58fcbd0e616734e2e230aaf4ab4;hp=1552096ef15aed6bf8f941a1c24474c9ca8cbe44;hpb=ea3673e89598b896c93ebee864e6cb8db7f6c6ec;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java index 1552096ef1..a4753a4fe4 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java @@ -97,9 +97,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { final Builder ftlBuilder = ImmutableMap.builder(); for (String followerId : context.getPeerAddresses().keySet()) { FollowerLogInformation followerLogInformation = - new FollowerLogInformationImpl(followerId, - context.getCommitIndex(), -1, - context.getConfigParams().getElectionTimeOutInterval()); + new FollowerLogInformationImpl(followerId, -1, context); ftlBuilder.put(followerId, followerLogInformation); } @@ -126,6 +124,9 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // (heartbeat) to each server; repeat during idle periods to // prevent election timeouts (§5.2) sendAppendEntries(0, false); + + // It is important to schedule this heartbeat here + scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); } /** @@ -157,8 +158,6 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if(LOG.isTraceEnabled()) { LOG.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply); - } else if(LOG.isDebugEnabled() && !appendEntriesReply.isSuccess()) { - LOG.debug("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply); } // Update the FollowerLogInformation @@ -173,7 +172,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if(followerLogInformation.timeSinceLastActivity() > context.getConfigParams().getElectionTimeOutInterval().toMillis()) { - LOG.error("{} : handleAppendEntriesReply delayed beyond election timeout, " + + LOG.warn("{} : handleAppendEntriesReply delayed beyond election timeout, " + "appendEntriesReply : {}, timeSinceLastActivity : {}, lastApplied : {}, commitIndex : {}", logName(), appendEntriesReply, followerLogInformation.timeSinceLastActivity(), context.getLastApplied(), context.getCommitIndex()); @@ -181,12 +180,17 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerLogInformation.markFollowerActive(); + boolean updated = false; if (appendEntriesReply.isSuccess()) { - followerLogInformation - .setMatchIndex(appendEntriesReply.getLogLastIndex()); - followerLogInformation - .setNextIndex(appendEntriesReply.getLogLastIndex() + 1); + updated = followerLogInformation.setMatchIndex(appendEntriesReply.getLogLastIndex()); + updated = followerLogInformation.setNextIndex(appendEntriesReply.getLogLastIndex() + 1) || updated; + + if(updated && LOG.isDebugEnabled()) { + LOG.debug("{}: handleAppendEntriesReply - FollowerLogInformation for {} updated: matchIndex: {}, nextIndex: {}", logName(), + followerId, followerLogInformation.getMatchIndex(), followerLogInformation.getNextIndex()); + } } else { + LOG.debug("{}: handleAppendEntriesReply: received unsuccessful reply: {}", logName(), appendEntriesReply); // TODO: When we find that the follower is out of sync with the // Leader we simply decrement that followers next index by 1. @@ -223,8 +227,10 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // Apply the change to the state machine if (context.getCommitIndex() > context.getLastApplied()) { - LOG.debug("{}: handleAppendEntriesReply: applying to log - commitIndex: {}, lastAppliedIndex: {}", - logName(), context.getCommitIndex(), context.getLastApplied()); + if(LOG.isDebugEnabled()) { + LOG.debug("{}: handleAppendEntriesReply from {}: applying to log - commitIndex: {}, lastAppliedIndex: {}", + logName(), followerId, context.getCommitIndex(), context.getLastApplied()); + } applyLogToStateMachine(context.getCommitIndex()); } @@ -234,7 +240,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } //Send the next log entry immediately, if possible, no need to wait for heartbeat to trigger that event - sendUpdatesToFollower(followerId, followerLogInformation, false, false); + sendUpdatesToFollower(followerId, followerLogInformation, false, !updated); return this; } @@ -304,28 +310,26 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } } - try { - if (message instanceof SendHeartBeat) { - beforeSendHeartbeat(); - sendHeartBeat(); - return this; + if (message instanceof SendHeartBeat) { + beforeSendHeartbeat(); + sendHeartBeat(); + scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); + return this; - } else if(message instanceof SendInstallSnapshot) { - // received from RaftActor - setSnapshot(Optional.of(((SendInstallSnapshot) message).getSnapshot())); - sendInstallSnapshot(); + } else if(message instanceof SendInstallSnapshot) { + // received from RaftActor + setSnapshot(Optional.of(((SendInstallSnapshot) message).getSnapshot())); + sendInstallSnapshot(); - } else if (message instanceof Replicate) { - replicate((Replicate) message); + } else if (message instanceof Replicate) { + replicate((Replicate) message); - } else if (message instanceof InstallSnapshotReply){ - handleInstallSnapshotReply((InstallSnapshotReply) message); + } else if (message instanceof InstallSnapshotReply){ + handleInstallSnapshotReply((InstallSnapshotReply) message); - } - } finally { - scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); } + return super.handleMessage(sender, message); } @@ -384,7 +388,10 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerToSnapshot.markSendStatus(false); } - if (!wasLastChunk && followerToSnapshot.canSendNextChunk()) { + if (wasLastChunk && !context.isSnapshotCaptureInitiated()) { + // Since the follower is now caught up try to purge the log. + purgeInMemoryLog(); + } else if (!wasLastChunk && followerToSnapshot.canSendNextChunk()) { ActorSelection followerActor = context.getPeerActorSelection(followerId); if(followerActor != null) { sendSnapshotChunk(followerActor, followerId); @@ -467,9 +474,9 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { long leaderLastIndex = context.getReplicatedLog().lastIndex(); long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex(); - if(!isHeartbeat || LOG.isTraceEnabled()) { - LOG.debug("{}: Checking sendAppendEntries for follower {}, leaderLastIndex: {}, leaderSnapShotIndex: {}", - logName(), followerId, leaderLastIndex, leaderSnapShotIndex); + if((!isHeartbeat && LOG.isDebugEnabled()) || LOG.isTraceEnabled()) { + LOG.debug("{}: Checking sendAppendEntries for follower {}, followerNextIndex {}, leaderLastIndex: {}, leaderSnapShotIndex: {}", + logName(), followerId, followerNextIndex, leaderLastIndex, leaderSnapShotIndex); } if (isFollowerActive && context.getReplicatedLog().isPresent(followerNextIndex)) { @@ -554,7 +561,6 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } else if (!context.isSnapshotCaptureInitiated()) { - LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", logName(), getLeaderId()); ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied()); long lastAppliedIndex = -1; long lastAppliedTerm = -1; @@ -570,10 +576,19 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { boolean isInstallSnapshotInitiated = true; long replicatedToAllIndex = super.getReplicatedToAllIndex(); ReplicatedLogEntry replicatedToAllEntry = context.getReplicatedLog().get(replicatedToAllIndex); - actor().tell(new CaptureSnapshot(lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm, - (replicatedToAllEntry != null ? replicatedToAllEntry.getIndex() : -1), - (replicatedToAllEntry != null ? replicatedToAllEntry.getTerm() : -1), - isInstallSnapshotInitiated), actor()); + + CaptureSnapshot captureSnapshot = new CaptureSnapshot( + lastIndex(), lastTerm(), lastAppliedIndex, lastAppliedTerm, + (replicatedToAllEntry != null ? replicatedToAllEntry.getIndex() : -1), + (replicatedToAllEntry != null ? replicatedToAllEntry.getTerm() : -1), + isInstallSnapshotInitiated); + + if(LOG.isDebugEnabled()) { + LOG.debug("{}: Initiating install snapshot to follower {}: {}", logName(), followerId, + captureSnapshot); + } + + actor().tell(captureSnapshot, actor()); context.setSnapshotCaptureInitiated(true); } } @@ -620,10 +635,12 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { ).toSerializable(), actor() ); - LOG.info("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", - logName(), followerActor.path(), - followerToSnapshot.getChunkIndex(), - followerToSnapshot.getTotalChunks()); + + if(LOG.isDebugEnabled()) { + LOG.debug("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}", + logName(), followerActor.path(), followerToSnapshot.getChunkIndex(), + followerToSnapshot.getTotalChunks()); + } } } catch (IOException e) { LOG.error("{}: InstallSnapshot failed for Leader.", logName(), e);