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 Cancellable newLeaderTimer;
  56     private final List<OnComplete> onCompleteCallbacks = new ArrayList<>();
  57     private long newLeaderTimeoutInMillis = 2000;
  58     private final Stopwatch transferTimer = Stopwatch.createUnstarted();
  59     private boolean isTransferring;
  60
  61     RaftActorLeadershipTransferCohort(RaftActor raftActor) {
  62         this.raftActor = raftActor;
  63     }
  64
  65     void init() {
  66         RaftActorContext context = raftActor.getRaftActorContext();
  67         RaftActorBehavior currentBehavior = raftActor.getCurrentBehavior();
  68
  69         transferTimer.start();
  70
  71         Optional<ActorRef> roleChangeNotifier = raftActor.getRoleChangeNotifier();
  72         if(roleChangeNotifier.isPresent()) {
  73             roleChangeNotifier.get().tell(raftActor.newLeaderStateChanged(context.getId(), null,
  74                     currentBehavior.getLeaderPayloadVersion()), raftActor.self());
  75         }
  76
  77         for(String peerId: context.getPeerIds()) {
  78             ActorSelection followerActor = context.getPeerActorSelection(peerId);
  79             if(followerActor != null) {
  80                 followerActor.tell(LeaderTransitioning.INSTANCE, context.getActor());
  81             }
  82         }
  83
  84         raftActor.pauseLeader(new TimedRunnable(context.getConfigParams().getElectionTimeOutInterval(), raftActor) {
  85             @Override
  86             protected void doRun() {
  87                 doTransfer();
  88             }
  89
  90             @Override
  91             protected void doCancel() {
  92                 LOG.debug("{}: pauseLeader timed out - aborting transfer", raftActor.persistenceId());
  93                 abortTransfer();
  94             }
  95         });
  96     }
  97
  98     /**
  99      * This method is invoked to perform the leadership transfer.
 100      */
 101     @VisibleForTesting
 102     void doTransfer() {
 103         RaftActorBehavior behavior = raftActor.getCurrentBehavior();
 104         // Sanity check...
 105         if(behavior instanceof Leader) {
 106             isTransferring = true;
 107             ((Leader)behavior).transferLeadership(this);
 108         } else {
 109             LOG.debug("{}: No longer the leader - skipping transfer", raftActor.persistenceId());
 110             finish(true);
 111         }
 112     }
 113
 114     /**
 115      * This method is invoked to abort leadership transfer on failure.
 116      */
 117     public void abortTransfer() {
 118         LOG.debug("{}: leader transfer aborted", raftActor.persistenceId());
 119         finish(false);
 120     }
 121
 122     /**
 123      * This method is invoked when leadership transfer was carried out and complete.
 124      */
 125     public void transferComplete() {
 126         LOG.debug("{}: leader transfer complete - waiting for new leader", raftActor.persistenceId());
 127
 128         // We'll give it a little time for the new leader to be elected to give the derived class a
 129         // chance to possibly complete work that was suspended while we were transferring. The
 130         // RequestVote message from the new leader candidate should cause us to step down as leader
 131         // and convert to follower due to higher term. We should then get an AppendEntries heart
 132         // beat with the new leader id.
 133
 134         // Add a timer in case we don't get a leader change - 2 sec should be plenty of time if a new
 135         // leader is elected. Note: the Runnable is sent as a message to the raftActor which executes it
 136         // safely run on the actor's thread dispatcher.
 137         FiniteDuration timeout = FiniteDuration.create(newLeaderTimeoutInMillis, TimeUnit.MILLISECONDS);
 138         newLeaderTimer = raftActor.getContext().system().scheduler().scheduleOnce(timeout, raftActor.self(),
 139                 new Runnable() {
 140                     @Override
 141                     public void run() {
 142                         LOG.debug("{}: leader not elected in time", raftActor.persistenceId());
 143                         finish(true);
 144                     }
 145                 }, raftActor.getContext().system().dispatcher(), raftActor.self());
 146     }
 147
 148     void onNewLeader(String newLeader) {
 149         if(newLeader != null && newLeaderTimer != null) {
 150             LOG.debug("{}: leader changed to {}", raftActor.persistenceId(), newLeader);
 151             newLeaderTimer.cancel();
 152             finish(true);
 153         }
 154     }
 155
 156     private void finish(boolean success) {
 157         isTransferring = false;
 158         if(transferTimer.isRunning()) {
 159             transferTimer.stop();
 160             if(success) {
 161                 LOG.info("{}: Successfully transferred leadership to {} in {}", raftActor.persistenceId(),
 162                         raftActor.getLeaderId(), transferTimer);
 163             } else {
 164                 LOG.warn("{}: Failed to transfer leadership in {}", raftActor.persistenceId(), transferTimer);
 165             }
 166         }
 167
 168         for(OnComplete onComplete: onCompleteCallbacks) {
 169             if(success) {
 170                 onComplete.onSuccess(raftActor.self());
 171             } else {
 172                 onComplete.onFailure(raftActor.self());
 173             }
 174         }
 175     }
 176
 177     void addOnComplete(OnComplete onComplete) {
 178         onCompleteCallbacks.add(onComplete);
 179     }
 180
 181     boolean isTransferring() {
 182         return isTransferring;
 183     }
 184
 185     @VisibleForTesting
 186     void setNewLeaderTimeoutInMillis(long newLeaderTimeoutInMillis) {
 187         this.newLeaderTimeoutInMillis = newLeaderTimeoutInMillis;
 188     }
 189
 190     interface OnComplete {
 191         void onSuccess(ActorRef raftActorRef);
 192         void onFailure(ActorRef raftActorRef);
 193     }
 194 }