opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorLeadershipTransferCohort.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.raft;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.ActorSelection;
  12 import akka.actor.Cancellable;
  13 import com.google.common.annotations.VisibleForTesting;
  14 import com.google.common.base.Optional;
  15 import com.google.common.base.Stopwatch;
  16 import java.util.ArrayList;
  17 import java.util.List;
  18 import java.util.concurrent.TimeUnit;
  19 import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning;
  20 import org.opendaylight.controller.cluster.raft.behaviors.Leader;
  21 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  22 import org.slf4j.Logger;
  23 import org.slf4j.LoggerFactory;
  24 import scala.concurrent.duration.FiniteDuration;
  25
  26 /**
  27  * A raft actor support class that participates in leadership transfer. An instance is created upon
  28  * initialization of leadership transfer.
  29  * <p>
  30  * The transfer process is as follows:
  31  * <ol>
  32  * <li>Send a LeaderStateChanged message with a null leader Id to the local RoleChangeNotifier to notify
  33  *     clients that we no longer have a working leader.</li>
  34  * <li>Send a LeaderTransitioning message to each follower so each can send LeaderStateChanged messages to
  35  *     their local RoleChangeNotifiers.</li>
  36  * <li>Call {@link RaftActor#pauseLeader} passing this RaftActorLeadershipTransferCohort
  37  *     instance. This allows derived classes to perform work prior to transferring leadership.</li>
  38  * <li>When the pause is complete, the {@link #run} method is called which in turn calls
  39  *     {@link Leader#transferLeadership}.</li>
  40  * <li>The Leader calls {@link #transferComplete} on successful completion.</li>
  41  * <li>Wait a short period of time for the new leader to be elected to give the derived class a chance to
  42  *     possibly complete work that was suspended while we were transferring.</li>
  43  * <li>On notification of the new leader from the RaftActor or on time out, notify {@link OnComplete} callbacks.</li>
  44  * </ol>
  45  * <p>
  46  * NOTE: All methods on this class must be called on the actor's thread dispatcher as they may access/modify
  47  * internal state.
  48  *
  49  * @author Thomas Pantelis
  50  */
  51 public class RaftActorLeadershipTransferCohort {
  52     private static final Logger LOG = LoggerFactory.getLogger(RaftActorLeadershipTransferCohort.class);
  53
  54     private final RaftActor raftActor;
  55     private final ActorRef replyTo;
  56     private Cancellable newLeaderTimer;
  57     private final List<OnComplete> onCompleteCallbacks = new ArrayList<>();
  58     private long newLeaderTimeoutInMillis = 2000;
  59     private final Stopwatch transferTimer = Stopwatch.createUnstarted();
  60     private boolean isTransferring;
  61
  62     RaftActorLeadershipTransferCohort(RaftActor raftActor, ActorRef replyTo) {
  63         this.raftActor = raftActor;
  64         this.replyTo = replyTo;
  65     }
  66
  67     void init() {
  68         RaftActorContext context = raftActor.getRaftActorContext();
  69         RaftActorBehavior currentBehavior = raftActor.getCurrentBehavior();
  70
  71         transferTimer.start();
  72
  73         Optional<ActorRef> roleChangeNotifier = raftActor.getRoleChangeNotifier();
  74         if(roleChangeNotifier.isPresent()) {
  75             roleChangeNotifier.get().tell(raftActor.newLeaderStateChanged(context.getId(), null,
  76                     currentBehavior.getLeaderPayloadVersion()), raftActor.self());
  77         }
  78
  79         for(String peerId: context.getPeerIds()) {
  80             ActorSelection followerActor = context.getPeerActorSelection(peerId);
  81             if(followerActor != null) {
  82                 followerActor.tell(LeaderTransitioning.INSTANCE, context.getActor());
  83             }
  84         }
  85
  86         raftActor.pauseLeader(new TimedRunnable(context.getConfigParams().getElectionTimeOutInterval(), raftActor) {
  87             @Override
  88             protected void doRun() {
  89                 doTransfer();
  90             }
  91
  92             @Override
  93             protected void doCancel() {
  94                 LOG.debug("{}: pauseLeader timed out - aborting transfer", raftActor.persistenceId());
  95                 abortTransfer();
  96             }
  97         });
  98     }
  99
 100     /**
 101      * This method is invoked to perform the leadership transfer.
 102      */
 103     @VisibleForTesting
 104     void doTransfer() {
 105         RaftActorBehavior behavior = raftActor.getCurrentBehavior();
 106         // Sanity check...
 107         if(behavior instanceof Leader) {
 108             isTransferring = true;
 109             ((Leader)behavior).transferLeadership(this);
 110         } else {
 111             LOG.debug("{}: No longer the leader - skipping transfer", raftActor.persistenceId());
 112             finish(true);
 113         }
 114     }
 115
 116     /**
 117      * This method is invoked to abort leadership transfer on failure.
 118      */
 119     public void abortTransfer() {
 120         LOG.debug("{}: leader transfer aborted", raftActor.persistenceId());
 121         finish(false);
 122     }
 123
 124     /**
 125      * This method is invoked when leadership transfer was carried out and complete.
 126      */
 127     public void transferComplete() {
 128         LOG.debug("{}: leader transfer complete - waiting for new leader", raftActor.persistenceId());
 129
 130         // We'll give it a little time for the new leader to be elected to give the derived class a
 131         // chance to possibly complete work that was suspended while we were transferring. The
 132         // RequestVote message from the new leader candidate should cause us to step down as leader
 133         // and convert to follower due to higher term. We should then get an AppendEntries heart
 134         // beat with the new leader id.
 135
 136         // Add a timer in case we don't get a leader change - 2 sec should be plenty of time if a new
 137         // leader is elected. Note: the Runnable is sent as a message to the raftActor which executes it
 138         // safely run on the actor's thread dispatcher.
 139         FiniteDuration timeout = FiniteDuration.create(newLeaderTimeoutInMillis, TimeUnit.MILLISECONDS);
 140         newLeaderTimer = raftActor.getContext().system().scheduler().scheduleOnce(timeout, raftActor.self(),
 141                 new Runnable() {
 142                     @Override
 143                     public void run() {
 144                         LOG.debug("{}: leader not elected in time", raftActor.persistenceId());
 145                         finish(true);
 146                     }
 147                 }, raftActor.getContext().system().dispatcher(), raftActor.self());
 148     }
 149
 150     void onNewLeader(String newLeader) {
 151         if(newLeader != null && newLeaderTimer != null) {
 152             LOG.debug("{}: leader changed to {}", raftActor.persistenceId(), newLeader);
 153             newLeaderTimer.cancel();
 154             finish(true);
 155         }
 156     }
 157
 158     private void finish(boolean success) {
 159         isTransferring = false;
 160         if(transferTimer.isRunning()) {
 161             transferTimer.stop();
 162             if(success) {
 163                 LOG.info("{}: Successfully transferred leadership to {} in {}", raftActor.persistenceId(),
 164                         raftActor.getLeaderId(), transferTimer.toString());
 165             } else {
 166                 LOG.warn("{}: Failed to transfer leadership in {}", raftActor.persistenceId(),
 167                         transferTimer.toString());
 168             }
 169         }
 170
 171         for(OnComplete onComplete: onCompleteCallbacks) {
 172             if(success) {
 173                 onComplete.onSuccess(raftActor.self(), replyTo);
 174             } else {
 175                 onComplete.onFailure(raftActor.self(), replyTo);
 176             }
 177         }
 178     }
 179
 180     void addOnComplete(OnComplete onComplete) {
 181         onCompleteCallbacks.add(onComplete);
 182     }
 183
 184     boolean isTransferring() {
 185         return isTransferring;
 186     }
 187
 188     @VisibleForTesting
 189     void setNewLeaderTimeoutInMillis(long newLeaderTimeoutInMillis) {
 190         this.newLeaderTimeoutInMillis = newLeaderTimeoutInMillis;
 191     }
 192
 193     interface OnComplete {
 194         void onSuccess(ActorRef raftActorRef, ActorRef replyTo);
 195         void onFailure(ActorRef raftActorRef, ActorRef replyTo);
 196     }
 197 }