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%2FLeader.java;h=924c30f657b927528a81ac9997564418c8ec2fe5;hb=af406fdd1a948e05014b95e2d23d59cda1ebd837;hp=ee3cc65dddb90815aecd50771c834d3ab461ecd4;hpb=989c3723e634bf4a0fb671246c31eadf255c462c;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java index ee3cc65ddd..924c30f657 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.java @@ -8,13 +8,20 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; -import akka.actor.Cancellable; +import akka.actor.ActorSelection; import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; +import com.google.common.base.Stopwatch; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.opendaylight.controller.cluster.raft.FollowerLogInformation; import org.opendaylight.controller.cluster.raft.RaftActorContext; -import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot; +import org.opendaylight.controller.cluster.raft.RaftActorLeadershipTransferCohort; +import org.opendaylight.controller.cluster.raft.RaftState; +import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.base.messages.IsolatedLeaderCheck; -import scala.concurrent.duration.FiniteDuration; +import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; /** * The behavior of a RaftActor when it is in the Leader state @@ -39,17 +46,13 @@ import scala.concurrent.duration.FiniteDuration; * set commitIndex = N (§5.3, §5.4). */ public class Leader extends AbstractLeader { - private Cancellable installSnapshotSchedule = null; - private Cancellable isolatedLeaderCheckSchedule = null; + private static final IsolatedLeaderCheck ISOLATED_LEADER_CHECK = new IsolatedLeaderCheck(); + private final Stopwatch isolatedLeaderCheck; + private @Nullable LeadershipTransferContext leadershipTransferContext; public Leader(RaftActorContext context) { super(context); - - scheduleInstallSnapshotCheck(context.getConfigParams().getIsolatedCheckInterval()); - - scheduleIsolatedLeaderCheck( - new FiniteDuration(context.getConfigParams().getHeartBeatInterval().length() * 10, - context.getConfigParams().getHeartBeatInterval().unit())); + isolatedLeaderCheck = Stopwatch.createStarted(); } @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) { @@ -57,55 +60,104 @@ public class Leader extends AbstractLeader { if (originalMessage instanceof IsolatedLeaderCheck) { if (isLeaderIsolated()) { - LOG.info("At least {} followers need to be active, Switching {} from Leader to IsolatedLeader", - minIsolatedLeaderPeerCount, leaderId); - return switchBehavior(new IsolatedLeader(context)); + LOG.warn("{}: At least {} followers need to be active, Switching {} from Leader to IsolatedLeader", + context.getId(), getMinIsolatedLeaderPeerCount(), leaderId); + + return internalSwitchBehavior(RaftState.IsolatedLeader); } } return super.handleMessage(sender, originalMessage); } - protected void stopInstallSnapshotSchedule() { - if (installSnapshotSchedule != null && !installSnapshotSchedule.isCancelled()) { - installSnapshotSchedule.cancel(); + @Override + protected void beforeSendHeartbeat(){ + if(isolatedLeaderCheck.elapsed(TimeUnit.MILLISECONDS) > context.getConfigParams().getIsolatedCheckIntervalInMillis()){ + context.getActor().tell(ISOLATED_LEADER_CHECK, context.getActor()); + isolatedLeaderCheck.reset().start(); } - } - protected void scheduleInstallSnapshotCheck(FiniteDuration interval) { - if (getFollowerIds().isEmpty()) { - // Optimization - do not bother scheduling a heartbeat as there are - // no followers - return; + if(leadershipTransferContext != null && leadershipTransferContext.isExpired( + context.getConfigParams().getElectionTimeOutInterval().toMillis())) { + LOG.debug("{}: Leadership transfer expired", logName()); + leadershipTransferContext = null; } + } - stopInstallSnapshotSchedule(); + @Override + protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply) { + RaftActorBehavior returnBehavior = super.handleAppendEntriesReply(sender, appendEntriesReply); + tryToCompleteLeadershipTransfer(appendEntriesReply.getFollowerId()); + return returnBehavior; + } - // Schedule a message to send append entries to followers that can - // accept an append entries with some data in it - installSnapshotSchedule = - context.getActorSystem().scheduler().scheduleOnce( - interval, - context.getActor(), new InitiateInstallSnapshot(), - context.getActorSystem().dispatcher(), context.getActor()); + /** + * Attempts to transfer leadership to a follower as per the raft paper (§3.10) as follows: + *