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

   1 /*
   2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
   3  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   4  *
   5  * This program and the accompanying materials are made available under the
   6  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   7  * and is available at http://www.eclipse.org/legal/epl-v10.html
   8  */
   9
  10 package org.opendaylight.controller.cluster.raft;
  11
  12 import akka.actor.ActorRef;
  13 import akka.actor.ActorSelection;
  14 import akka.actor.PoisonPill;
  15 import akka.japi.Procedure;
  16 import com.google.common.annotations.VisibleForTesting;
  17 import com.google.common.base.Optional;
  18 import com.google.common.collect.Lists;
  19 import java.io.Serializable;
  20 import java.util.Collection;
  21 import java.util.HashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.Objects;
  25 import java.util.concurrent.TimeUnit;
  26 import javax.annotation.Nonnull;
  27 import org.apache.commons.lang3.time.DurationFormatUtils;
  28 import org.opendaylight.controller.cluster.DataPersistenceProvider;
  29 import org.opendaylight.controller.cluster.DelegatingPersistentDataProvider;
  30 import org.opendaylight.controller.cluster.NonPersistentDataProvider;
  31 import org.opendaylight.controller.cluster.PersistentDataProvider;
  32 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActor;
  33 import org.opendaylight.controller.cluster.notifications.LeaderStateChanged;
  34 import org.opendaylight.controller.cluster.notifications.RoleChanged;
  35 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
  36 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  37 import org.opendaylight.controller.cluster.raft.base.messages.InitiateCaptureSnapshot;
  38 import org.opendaylight.controller.cluster.raft.base.messages.LeaderTransitioning;
  39 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
  40 import org.opendaylight.controller.cluster.raft.base.messages.SwitchBehavior;
  41 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
  42 import org.opendaylight.controller.cluster.raft.behaviors.AbstractRaftActorBehavior;
  43 import org.opendaylight.controller.cluster.raft.behaviors.Follower;
  44 import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
  45 import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
  46 import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
  47 import org.opendaylight.controller.cluster.raft.client.messages.FollowerInfo;
  48 import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState;
  49 import org.opendaylight.controller.cluster.raft.client.messages.OnDemandRaftState;
  50 import org.opendaylight.controller.cluster.raft.client.messages.Shutdown;
  51 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 /**
  56  * RaftActor encapsulates a state machine that needs to be kept synchronized
  57  * in a cluster. It implements the RAFT algorithm as described in the paper
  58  * <a href='https://ramcloud.stanford.edu/wiki/download/attachments/11370504/raft.pdf'>
  59  * In Search of an Understandable Consensus Algorithm</a>
  60  * <p/>
  61  * RaftActor has 3 states and each state has a certain behavior associated
  62  * with it. A Raft actor can behave as,
  63  * <ul>
  64  * <li> A Leader </li>
  65  * <li> A Follower (or) </li>
  66  * <li> A Candidate </li>
  67  * </ul>
  68  * <p/>
  69  * <p/>
  70  * A RaftActor MUST be a Leader in order to accept requests from clients to
  71  * change the state of it's encapsulated state machine. Once a RaftActor becomes
  72  * a Leader it is also responsible for ensuring that all followers ultimately
  73  * have the same log and therefore the same state machine as itself.
  74  * <p/>
  75  * <p/>
  76  * The current behavior of a RaftActor determines how election for leadership
  77  * is initiated and how peer RaftActors react to request for votes.
  78  * <p/>
  79  * <p/>
  80  * Each RaftActor also needs to know the current election term. It uses this
  81  * information for a couple of things. One is to simply figure out who it
  82  * voted for in the last election. Another is to figure out if the message
  83  * it received to update it's state is stale.
  84  * <p/>
  85  * <p/>
  86  * The RaftActor uses akka-persistence to store it's replicated log.
  87  * Furthermore through it's behaviors a Raft Actor determines
  88  * <p/>
  89  * <ul>
  90  * <li> when a log entry should be persisted </li>
  91  * <li> when a log entry should be applied to the state machine (and) </li>
  92  * <li> when a snapshot should be saved </li>
  93  * </ul>
  94  */
  95 public abstract class RaftActor extends AbstractUntypedPersistentActor {
  96
  97     private static final long APPLY_STATE_DELAY_THRESHOLD_IN_NANOS = TimeUnit.MILLISECONDS.toNanos(50L); // 50 millis
  98
  99     protected final Logger LOG = LoggerFactory.getLogger(getClass());
 100
 101     /**
 102      * This context should NOT be passed directly to any other actor it is
 103      * only to be consumed by the RaftActorBehaviors
 104      */
 105     private final RaftActorContextImpl context;
 106
 107     private final DelegatingPersistentDataProvider delegatingPersistenceProvider;
 108
 109     private final PersistentDataProvider persistentProvider;
 110
 111     private RaftActorRecoverySupport raftRecovery;
 112
 113     private RaftActorSnapshotMessageSupport snapshotSupport;
 114
 115     private final BehaviorStateHolder reusableBehaviorStateHolder = new BehaviorStateHolder();
 116
 117     private RaftActorServerConfigurationSupport serverConfigurationSupport;
 118
 119     private RaftActorLeadershipTransferCohort leadershipTransferInProgress;
 120
 121     private boolean shuttingDown;
 122
 123     public RaftActor(String id, Map<String, String> peerAddresses,
 124          Optional<ConfigParams> configParams, short payloadVersion) {
 125
 126         persistentProvider = new PersistentDataProvider(this);
 127         delegatingPersistenceProvider = new RaftActorDelegatingPersistentDataProvider(null, persistentProvider);
 128
 129         context = new RaftActorContextImpl(this.getSelf(),
 130             this.getContext(), id, new ElectionTermImpl(persistentProvider, id, LOG),
 131             -1, -1, peerAddresses,
 132             (configParams.isPresent() ? configParams.get(): new DefaultConfigParamsImpl()),
 133             delegatingPersistenceProvider, LOG);
 134
 135         context.setPayloadVersion(payloadVersion);
 136         context.setReplicatedLog(ReplicatedLogImpl.newInstance(context));
 137     }
 138
 139     @Override
 140     public void preStart() throws Exception {
 141         LOG.info("Starting recovery for {} with journal batch size {}", persistenceId(),
 142                 context.getConfigParams().getJournalRecoveryLogBatchSize());
 143
 144         super.preStart();
 145
 146         snapshotSupport = newRaftActorSnapshotMessageSupport();
 147         serverConfigurationSupport = new RaftActorServerConfigurationSupport(this);
 148     }
 149
 150     @Override
 151     public void postStop() {
 152         context.close();
 153         super.postStop();
 154     }
 155
 156     @Override
 157     protected void handleRecover(Object message) {
 158         if(raftRecovery == null) {
 159             raftRecovery = newRaftActorRecoverySupport();
 160         }
 161
 162         boolean recoveryComplete = raftRecovery.handleRecoveryMessage(message, persistentProvider);
 163         if(recoveryComplete) {
 164             onRecoveryComplete();
 165
 166             initializeBehavior();
 167
 168             raftRecovery = null;
 169
 170             if (context.getReplicatedLog().size() > 0) {
 171                 self().tell(new InitiateCaptureSnapshot(), self());
 172                 LOG.info("{}: Snapshot capture initiated after recovery", persistenceId());
 173             } else {
 174                 LOG.info("{}: Snapshot capture NOT initiated after recovery, journal empty", persistenceId());
 175             }
 176         }
 177     }
 178
 179     protected RaftActorRecoverySupport newRaftActorRecoverySupport() {
 180         return new RaftActorRecoverySupport(context, getRaftActorRecoveryCohort());
 181     }
 182
 183     @VisibleForTesting
 184     void initializeBehavior(){
 185         changeCurrentBehavior(new Follower(context));
 186     }
 187
 188     @VisibleForTesting
 189     protected void changeCurrentBehavior(RaftActorBehavior newBehavior) {
 190         final RaftActorBehavior currentBehavior = getCurrentBehavior();
 191         if (currentBehavior != null) {
 192             try {
 193                 currentBehavior.close();
 194             } catch (Exception e) {
 195                 LOG.warn("{}: Error closing behavior {}", persistence(), currentBehavior, e);
 196             }
 197         }
 198
 199         reusableBehaviorStateHolder.init(currentBehavior);
 200         setCurrentBehavior(newBehavior);
 201         handleBehaviorChange(reusableBehaviorStateHolder, newBehavior);
 202     }
 203
 204     @Override
 205     protected void handleCommand(final Object message) {
 206         if (serverConfigurationSupport.handleMessage(message, getSender())) {
 207             return;
 208         }
 209         if (snapshotSupport.handleSnapshotMessage(message, getSender())) {
 210             return;
 211         }
 212
 213         if (message instanceof ApplyState) {
 214             ApplyState applyState = (ApplyState) message;
 215
 216             long elapsedTime = (System.nanoTime() - applyState.getStartTime());
 217             if(elapsedTime >= APPLY_STATE_DELAY_THRESHOLD_IN_NANOS){
 218                 LOG.warn("ApplyState took more time than expected. Elapsed Time = {} ms ApplyState = {}",
 219                         TimeUnit.NANOSECONDS.toMillis(elapsedTime), applyState);
 220             }
 221
 222             if(LOG.isDebugEnabled()) {
 223                 LOG.debug("{}: Applying state for log index {} data {}",
 224                     persistenceId(), applyState.getReplicatedLogEntry().getIndex(),
 225                     applyState.getReplicatedLogEntry().getData());
 226             }
 227
 228             applyState(applyState.getClientActor(), applyState.getIdentifier(),
 229                 applyState.getReplicatedLogEntry().getData());
 230
 231             if (!hasFollowers()) {
 232                 // for single node, the capture should happen after the apply state
 233                 // as we delete messages from the persistent journal which have made it to the snapshot
 234                 // capturing the snapshot before applying makes the persistent journal and snapshot out of sync
 235                 // and recovery shows data missing
 236                 context.getReplicatedLog().captureSnapshotIfReady(applyState.getReplicatedLogEntry());
 237
 238                 context.getSnapshotManager().trimLog(context.getLastApplied());
 239             }
 240
 241         } else if (message instanceof ApplyJournalEntries) {
 242             ApplyJournalEntries applyEntries = (ApplyJournalEntries) message;
 243             if(LOG.isDebugEnabled()) {
 244                 LOG.debug("{}: Persisting ApplyJournalEntries with index={}", persistenceId(), applyEntries.getToIndex());
 245             }
 246
 247             persistence().persist(applyEntries, NoopProcedure.instance());
 248
 249         } else if (message instanceof FindLeader) {
 250             getSender().tell(
 251                 new FindLeaderReply(getLeaderAddress()),
 252                 getSelf()
 253             );
 254         } else if(message instanceof GetOnDemandRaftState) {
 255             onGetOnDemandRaftStats();
 256         } else if(message instanceof InitiateCaptureSnapshot) {
 257             captureSnapshot();
 258         } else if(message instanceof SwitchBehavior) {
 259             switchBehavior(((SwitchBehavior) message));
 260         } else if(message instanceof LeaderTransitioning) {
 261             onLeaderTransitioning();
 262         } else if(message instanceof Shutdown) {
 263             onShutDown();
 264         } else if(message instanceof Runnable) {
 265             ((Runnable)message).run();
 266         } else {
 267             final RaftActorBehavior currentBehavior = getCurrentBehavior();
 268
 269             // Processing the message may affect the state, hence we need to capture it
 270             reusableBehaviorStateHolder.init(currentBehavior);
 271             final RaftActorBehavior nextBehavior = currentBehavior.handleMessage(getSender(), message);
 272             switchBehavior(nextBehavior);
 273         }
 274     }
 275
 276     private void initiateLeadershipTransfer(final RaftActorLeadershipTransferCohort.OnComplete onComplete) {
 277         LOG.debug("{}: Initiating leader transfer", persistenceId());
 278
 279         if(leadershipTransferInProgress == null) {
 280             leadershipTransferInProgress = new RaftActorLeadershipTransferCohort(this);
 281             leadershipTransferInProgress.addOnComplete(new RaftActorLeadershipTransferCohort.OnComplete() {
 282                 @Override
 283                 public void onSuccess(ActorRef raftActorRef) {
 284                     leadershipTransferInProgress = null;
 285                 }
 286
 287                 @Override
 288                 public void onFailure(ActorRef raftActorRef) {
 289                     leadershipTransferInProgress = null;
 290                 }
 291             });
 292
 293             leadershipTransferInProgress.addOnComplete(onComplete);
 294             leadershipTransferInProgress.init();
 295         } else {
 296             LOG.debug("{}: prior leader transfer in progress - adding callback", persistenceId());
 297             leadershipTransferInProgress.addOnComplete(onComplete);
 298         }
 299     }
 300
 301     private void onShutDown() {
 302         LOG.debug("{}: onShutDown", persistenceId());
 303
 304         if(shuttingDown) {
 305             return;
 306         }
 307
 308         shuttingDown = true;
 309
 310         final RaftActorBehavior currentBehavior = context.getCurrentBehavior();
 311         if (currentBehavior.state() != RaftState.Leader) {
 312             // For non-leaders shutdown is a no-op
 313             self().tell(PoisonPill.getInstance(), self());
 314             return;
 315         }
 316
 317         if (context.hasFollowers()) {
 318             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
 319                 @Override
 320                 public void onSuccess(ActorRef raftActorRef) {
 321                     LOG.debug("{}: leader transfer succeeded - sending PoisonPill", persistenceId());
 322                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
 323                 }
 324
 325                 @Override
 326                 public void onFailure(ActorRef raftActorRef) {
 327                     LOG.debug("{}: leader transfer failed - sending PoisonPill", persistenceId());
 328                     raftActorRef.tell(PoisonPill.getInstance(), raftActorRef);
 329                 }
 330             });
 331         } else {
 332             pauseLeader(new TimedRunnable(context.getConfigParams().getElectionTimeOutInterval(), this) {
 333                 @Override
 334                 protected void doRun() {
 335                     self().tell(PoisonPill.getInstance(), self());
 336                 }
 337
 338                 @Override
 339                 protected void doCancel() {
 340                     self().tell(PoisonPill.getInstance(), self());
 341                 }
 342             });
 343         }
 344     }
 345
 346     private void onLeaderTransitioning() {
 347         LOG.debug("{}: onLeaderTransitioning", persistenceId());
 348         Optional<ActorRef> roleChangeNotifier = getRoleChangeNotifier();
 349         if(getRaftState() == RaftState.Follower && roleChangeNotifier.isPresent()) {
 350             roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), null,
 351                 getCurrentBehavior().getLeaderPayloadVersion()), getSelf());
 352         }
 353     }
 354
 355     private void switchBehavior(SwitchBehavior message) {
 356         if(!getRaftActorContext().getRaftPolicy().automaticElectionsEnabled()) {
 357             RaftState newState = message.getNewState();
 358             if( newState == RaftState.Leader || newState == RaftState.Follower) {
 359                 reusableBehaviorStateHolder.init(getCurrentBehavior());
 360                 final RaftActorBehavior nextBehavior = AbstractRaftActorBehavior.createBehavior(context,
 361                     message.getNewState());
 362                 switchBehavior(nextBehavior);
 363                 getRaftActorContext().getTermInformation().updateAndPersist(message.getNewTerm(), "");
 364             } else {
 365                 LOG.warn("Switching to behavior : {} - not supported", newState);
 366             }
 367         }
 368     }
 369
 370     private void switchBehavior(final RaftActorBehavior nextBehavior) {
 371         setCurrentBehavior(nextBehavior);
 372         handleBehaviorChange(reusableBehaviorStateHolder, nextBehavior);
 373     }
 374
 375     @VisibleForTesting
 376     RaftActorSnapshotMessageSupport newRaftActorSnapshotMessageSupport() {
 377         return new RaftActorSnapshotMessageSupport(context, getRaftActorSnapshotCohort());
 378     }
 379
 380     private void onGetOnDemandRaftStats() {
 381         // Debugging message to retrieve raft stats.
 382
 383         Map<String, String> peerAddresses = new HashMap<>();
 384         for(String peerId: context.getPeerIds()) {
 385             peerAddresses.put(peerId, context.getPeerAddress(peerId));
 386         }
 387
 388         final RaftActorBehavior currentBehavior = context.getCurrentBehavior();
 389         OnDemandRaftState.Builder builder = OnDemandRaftState.builder()
 390                 .commitIndex(context.getCommitIndex())
 391                 .currentTerm(context.getTermInformation().getCurrentTerm())
 392                 .inMemoryJournalDataSize(replicatedLog().dataSize())
 393                 .inMemoryJournalLogSize(replicatedLog().size())
 394                 .isSnapshotCaptureInitiated(context.getSnapshotManager().isCapturing())
 395                 .lastApplied(context.getLastApplied())
 396                 .lastIndex(replicatedLog().lastIndex())
 397                 .lastTerm(replicatedLog().lastTerm())
 398                 .leader(getLeaderId())
 399                 .raftState(currentBehavior.state().toString())
 400                 .replicatedToAllIndex(currentBehavior.getReplicatedToAllIndex())
 401                 .snapshotIndex(replicatedLog().getSnapshotIndex())
 402                 .snapshotTerm(replicatedLog().getSnapshotTerm())
 403                 .votedFor(context.getTermInformation().getVotedFor())
 404                 .peerAddresses(peerAddresses)
 405                 .customRaftPolicyClassName(context.getConfigParams().getCustomRaftPolicyImplementationClass());
 406
 407         ReplicatedLogEntry lastLogEntry = replicatedLog().last();
 408         if (lastLogEntry != null) {
 409             builder.lastLogIndex(lastLogEntry.getIndex());
 410             builder.lastLogTerm(lastLogEntry.getTerm());
 411         }
 412
 413         if(getCurrentBehavior() instanceof AbstractLeader) {
 414             AbstractLeader leader = (AbstractLeader)getCurrentBehavior();
 415             Collection<String> followerIds = leader.getFollowerIds();
 416             List<FollowerInfo> followerInfoList = Lists.newArrayListWithCapacity(followerIds.size());
 417             for(String id: followerIds) {
 418                 final FollowerLogInformation info = leader.getFollower(id);
 419                 followerInfoList.add(new FollowerInfo(id, info.getNextIndex(), info.getMatchIndex(),
 420                         info.isFollowerActive(), DurationFormatUtils.formatDurationHMS(info.timeSinceLastActivity())));
 421             }
 422
 423             builder.followerInfoList(followerInfoList);
 424         }
 425
 426         sender().tell(builder.build(), self());
 427
 428     }
 429
 430     private void handleBehaviorChange(BehaviorStateHolder oldBehaviorState, RaftActorBehavior currentBehavior) {
 431         RaftActorBehavior oldBehavior = oldBehaviorState.getBehavior();
 432
 433         if (oldBehavior != currentBehavior){
 434             onStateChanged();
 435         }
 436
 437         String lastValidLeaderId = oldBehavior == null ? null : oldBehaviorState.getLastValidLeaderId();
 438         String oldBehaviorStateName = oldBehavior == null ? null : oldBehavior.state().name();
 439
 440         // it can happen that the state has not changed but the leader has changed.
 441         Optional<ActorRef> roleChangeNotifier = getRoleChangeNotifier();
 442         if(!Objects.equals(lastValidLeaderId, currentBehavior.getLeaderId()) ||
 443            oldBehaviorState.getLeaderPayloadVersion() != currentBehavior.getLeaderPayloadVersion()) {
 444             if(roleChangeNotifier.isPresent()) {
 445                 roleChangeNotifier.get().tell(newLeaderStateChanged(getId(), currentBehavior.getLeaderId(),
 446                         currentBehavior.getLeaderPayloadVersion()), getSelf());
 447             }
 448
 449             onLeaderChanged(lastValidLeaderId, currentBehavior.getLeaderId());
 450
 451             if(leadershipTransferInProgress != null) {
 452                 leadershipTransferInProgress.onNewLeader(currentBehavior.getLeaderId());
 453             }
 454
 455             serverConfigurationSupport.onNewLeader(currentBehavior.getLeaderId());
 456         }
 457
 458         if (roleChangeNotifier.isPresent() &&
 459                 (oldBehavior == null || (oldBehavior.state() != currentBehavior.state()))) {
 460             roleChangeNotifier.get().tell(new RoleChanged(getId(), oldBehaviorStateName ,
 461                     currentBehavior.state().name()), getSelf());
 462         }
 463     }
 464
 465     protected LeaderStateChanged newLeaderStateChanged(String memberId, String leaderId, short leaderPayloadVersion) {
 466         return new LeaderStateChanged(memberId, leaderId, leaderPayloadVersion);
 467     }
 468
 469     @Override
 470     public long snapshotSequenceNr() {
 471         // When we do a snapshot capture, we also capture and save the sequence-number of the persistent journal,
 472         // so that we can delete the persistent journal based on the saved sequence-number
 473         // However , when akka replays the journal during recovery, it replays it from the sequence number when the snapshot
 474         // was saved and not the number we saved.
 475         // We would want to override it , by asking akka to use the last-sequence number known to us.
 476         return context.getSnapshotManager().getLastSequenceNumber();
 477     }
 478
 479     /**
 480      * When a derived RaftActor needs to persist something it must call
 481      * persistData.
 482      *
 483      * @param clientActor
 484      * @param identifier
 485      * @param data
 486      */
 487     protected void persistData(final ActorRef clientActor, final String identifier,
 488         final Payload data) {
 489
 490         ReplicatedLogEntry replicatedLogEntry = new ReplicatedLogImplEntry(
 491             context.getReplicatedLog().lastIndex() + 1,
 492             context.getTermInformation().getCurrentTerm(), data);
 493
 494         if(LOG.isDebugEnabled()) {
 495             LOG.debug("{}: Persist data {}", persistenceId(), replicatedLogEntry);
 496         }
 497
 498         final RaftActorContext raftContext = getRaftActorContext();
 499
 500         replicatedLog().appendAndPersist(replicatedLogEntry, new Procedure<ReplicatedLogEntry>() {
 501             @Override
 502             public void apply(ReplicatedLogEntry replicatedLogEntry) {
 503                 if (!hasFollowers()){
 504                     // Increment the Commit Index and the Last Applied values
 505                     raftContext.setCommitIndex(replicatedLogEntry.getIndex());
 506                     raftContext.setLastApplied(replicatedLogEntry.getIndex());
 507
 508                     // Apply the state immediately.
 509                     self().tell(new ApplyState(clientActor, identifier, replicatedLogEntry), self());
 510
 511                     // Send a ApplyJournalEntries message so that we write the fact that we applied
 512                     // the state to durable storage
 513                     self().tell(new ApplyJournalEntries(replicatedLogEntry.getIndex()), self());
 514
 515                 } else if (clientActor != null) {
 516                     context.getReplicatedLog().captureSnapshotIfReady(replicatedLogEntry);
 517
 518                     // Send message for replication
 519                     getCurrentBehavior().handleMessage(getSelf(),
 520                             new Replicate(clientActor, identifier, replicatedLogEntry));
 521                 }
 522             }
 523         });
 524     }
 525
 526     private ReplicatedLog replicatedLog() {
 527         return context.getReplicatedLog();
 528     }
 529
 530     protected String getId() {
 531         return context.getId();
 532     }
 533
 534     @VisibleForTesting
 535     void setCurrentBehavior(RaftActorBehavior behavior) {
 536         context.setCurrentBehavior(behavior);
 537     }
 538
 539     protected RaftActorBehavior getCurrentBehavior() {
 540         return context.getCurrentBehavior();
 541     }
 542
 543     /**
 544      * Derived actors can call the isLeader method to check if the current
 545      * RaftActor is the Leader or not
 546      *
 547      * @return true it this RaftActor is a Leader false otherwise
 548      */
 549     protected boolean isLeader() {
 550         return context.getId().equals(getCurrentBehavior().getLeaderId());
 551     }
 552
 553     protected final boolean isLeaderActive() {
 554         return getRaftState() != RaftState.IsolatedLeader && !shuttingDown &&
 555                 !isLeadershipTransferInProgress();
 556     }
 557
 558     private boolean isLeadershipTransferInProgress() {
 559         return leadershipTransferInProgress != null && leadershipTransferInProgress.isTransferring();
 560     }
 561
 562     /**
 563      * Derived actor can call getLeader if they need a reference to the Leader.
 564      * This would be useful for example in forwarding a request to an actor
 565      * which is the leader
 566      *
 567      * @return A reference to the leader if known, null otherwise
 568      */
 569     protected ActorSelection getLeader(){
 570         String leaderAddress = getLeaderAddress();
 571
 572         if(leaderAddress == null){
 573             return null;
 574         }
 575
 576         return context.actorSelection(leaderAddress);
 577     }
 578
 579     /**
 580      *
 581      * @return the current leader's id
 582      */
 583     protected final String getLeaderId(){
 584         return getCurrentBehavior().getLeaderId();
 585     }
 586
 587     @VisibleForTesting
 588     protected final RaftState getRaftState() {
 589         return getCurrentBehavior().state();
 590     }
 591
 592     protected Long getCurrentTerm(){
 593         return context.getTermInformation().getCurrentTerm();
 594     }
 595
 596     protected RaftActorContext getRaftActorContext() {
 597         return context;
 598     }
 599
 600     protected void updateConfigParams(ConfigParams configParams) {
 601
 602         // obtain the RaftPolicy for oldConfigParams and the updated one.
 603         String oldRaftPolicy = context.getConfigParams().
 604             getCustomRaftPolicyImplementationClass();
 605         String newRaftPolicy = configParams.
 606             getCustomRaftPolicyImplementationClass();
 607
 608         LOG.debug("{}: RaftPolicy used with prev.config {}, RaftPolicy used with newConfig {}", persistenceId(),
 609             oldRaftPolicy, newRaftPolicy);
 610         context.setConfigParams(configParams);
 611         if (!Objects.equals(oldRaftPolicy, newRaftPolicy)) {
 612             // The RaftPolicy was modified. If the current behavior is Follower then re-initialize to Follower
 613             // but transfer the previous leaderId so it doesn't immediately try to schedule an election. This
 614             // avoids potential disruption. Otherwise, switch to Follower normally.
 615             RaftActorBehavior behavior = getCurrentBehavior();
 616             if (behavior != null && behavior.state() == RaftState.Follower) {
 617                 String previousLeaderId = behavior.getLeaderId();
 618                 short previousLeaderPayloadVersion = behavior.getLeaderPayloadVersion();
 619
 620                 LOG.debug("{}: Re-initializing to Follower with previous leaderId {}", persistenceId(), previousLeaderId);
 621
 622                 changeCurrentBehavior(new Follower(context, previousLeaderId, previousLeaderPayloadVersion));
 623             } else {
 624                 initializeBehavior();
 625             }
 626         }
 627     }
 628
 629     public final DataPersistenceProvider persistence() {
 630         return delegatingPersistenceProvider.getDelegate();
 631     }
 632
 633     public void setPersistence(DataPersistenceProvider provider) {
 634         delegatingPersistenceProvider.setDelegate(provider);
 635     }
 636
 637     protected void setPersistence(boolean persistent) {
 638         if(persistent) {
 639             setPersistence(new PersistentDataProvider(this));
 640         } else {
 641             setPersistence(new NonPersistentDataProvider() {
 642                 /**
 643                  * The way snapshotting works is,
 644                  * <ol>
 645                  * <li> RaftActor calls createSnapshot on the Shard
 646                  * <li> Shard sends a CaptureSnapshotReply and RaftActor then calls saveSnapshot
 647                  * <li> When saveSnapshot is invoked on the akka-persistence API it uses the SnapshotStore to save
 648                  * the snapshot. The SnapshotStore sends SaveSnapshotSuccess or SaveSnapshotFailure. When the
 649                  * RaftActor gets SaveSnapshot success it commits the snapshot to the in-memory journal. This
 650                  * commitSnapshot is mimicking what is done in SaveSnapshotSuccess.
 651                  * </ol>
 652                  */
 653                 @Override
 654                 public void saveSnapshot(Object o) {
 655                     // Make saving Snapshot successful
 656                     // Committing the snapshot here would end up calling commit in the creating state which would
 657                     // be a state violation. That's why now we send a message to commit the snapshot.
 658                     self().tell(RaftActorSnapshotMessageSupport.COMMIT_SNAPSHOT, self());
 659                 }
 660             });
 661         }
 662     }
 663
 664     /**
 665      * setPeerAddress sets the address of a known peer at a later time.
 666      * <p>
 667      * This is to account for situations where a we know that a peer
 668      * exists but we do not know an address up-front. This may also be used in
 669      * situations where a known peer starts off in a different location and we
 670      * need to change it's address
 671      * <p>
 672      * Note that if the peerId does not match the list of peers passed to
 673      * this actor during construction an IllegalStateException will be thrown.
 674      *
 675      * @param peerId
 676      * @param peerAddress
 677      */
 678     protected void setPeerAddress(String peerId, String peerAddress){
 679         context.setPeerAddress(peerId, peerAddress);
 680     }
 681
 682     /**
 683      * The applyState method will be called by the RaftActor when some data
 684      * needs to be applied to the actor's state
 685      *
 686      * @param clientActor A reference to the client who sent this message. This
 687      *                    is the same reference that was passed to persistData
 688      *                    by the derived actor. clientActor may be null when
 689      *                    the RaftActor is behaving as a follower or during
 690      *                    recovery.
 691      * @param identifier  The identifier of the persisted data. This is also
 692      *                    the same identifier that was passed to persistData by
 693      *                    the derived actor. identifier may be null when
 694      *                    the RaftActor is behaving as a follower or during
 695      *                    recovery
 696      * @param data        A piece of data that was persisted by the persistData call.
 697      *                    This should NEVER be null.
 698      */
 699     protected abstract void applyState(ActorRef clientActor, String identifier,
 700         Object data);
 701
 702     /**
 703      * Returns the RaftActorRecoveryCohort to participate in persistence recovery.
 704      */
 705     @Nonnull
 706     protected abstract RaftActorRecoveryCohort getRaftActorRecoveryCohort();
 707
 708     /**
 709      * This method is called when recovery is complete.
 710      */
 711     protected abstract void onRecoveryComplete();
 712
 713     /**
 714      * Returns the RaftActorSnapshotCohort to participate in persistence recovery.
 715      */
 716     @Nonnull
 717     protected abstract RaftActorSnapshotCohort getRaftActorSnapshotCohort();
 718
 719     /**
 720      * This method will be called by the RaftActor when the state of the
 721      * RaftActor changes. The derived actor can then use methods like
 722      * isLeader or getLeader to do something useful
 723      */
 724     protected abstract void onStateChanged();
 725
 726     /**
 727      * Notifier Actor for this RaftActor to notify when a role change happens
 728      * @return ActorRef - ActorRef of the notifier or Optional.absent if none.
 729      */
 730     protected abstract Optional<ActorRef> getRoleChangeNotifier();
 731
 732     /**
 733      * This method is called prior to operations such as leadership transfer and actor shutdown when the leader
 734      * must pause or stop its duties. This method allows derived classes to gracefully pause or finish current
 735      * work prior to performing the operation. On completion of any work, the run method must be called on the
 736      * given Runnable to proceed with the given operation. <b>Important:</b> the run method must be called on
 737      * this actor's thread dispatcher as as it modifies internal state.
 738      * <p>
 739      * The default implementation immediately runs the operation.
 740      *
 741      * @param operation the operation to run
 742      */
 743     protected void pauseLeader(Runnable operation) {
 744         operation.run();
 745     }
 746
 747     protected void onLeaderChanged(String oldLeader, String newLeader) {
 748
 749     };
 750
 751     private String getLeaderAddress(){
 752         if(isLeader()){
 753             return getSelf().path().toString();
 754         }
 755         String leaderId = getLeaderId();
 756         if (leaderId == null) {
 757             return null;
 758         }
 759         String peerAddress = context.getPeerAddress(leaderId);
 760         if(LOG.isDebugEnabled()) {
 761             LOG.debug("{}: getLeaderAddress leaderId = {} peerAddress = {}",
 762                     persistenceId(), leaderId, peerAddress);
 763         }
 764
 765         return peerAddress;
 766     }
 767
 768     protected boolean hasFollowers(){
 769         return getRaftActorContext().hasFollowers();
 770     }
 771
 772     private void captureSnapshot() {
 773         SnapshotManager snapshotManager = context.getSnapshotManager();
 774
 775         if (!snapshotManager.isCapturing()) {
 776             final long idx = getCurrentBehavior().getReplicatedToAllIndex();
 777             LOG.debug("Take a snapshot of current state. lastReplicatedLog is {} and replicatedToAllIndex is {}",
 778                 replicatedLog().last(), idx);
 779
 780             snapshotManager.capture(replicatedLog().last(), idx);
 781         }
 782     }
 783
 784     /**
 785      * Switch this member to non-voting status. This is a no-op for all behaviors except when we are the leader,
 786      * in which case we need to step down.
 787      */
 788     void becomeNonVoting() {
 789         if (isLeader()) {
 790             initiateLeadershipTransfer(new RaftActorLeadershipTransferCohort.OnComplete() {
 791                 @Override
 792                 public void onSuccess(ActorRef raftActorRef) {
 793                     LOG.debug("{}: leader transfer succeeded after change to non-voting", persistenceId());
 794                     ensureFollowerState();
 795                 }
 796
 797                 @Override
 798                 public void onFailure(ActorRef raftActorRef) {
 799                     LOG.debug("{}: leader transfer failed after change to non-voting", persistenceId());
 800                     ensureFollowerState();
 801                 }
 802
 803                 private void ensureFollowerState() {
 804                     // Whether or not leadership transfer succeeded, we have to step down as leader and
 805                     // switch to Follower so ensure that.
 806                     if (getRaftState() != RaftState.Follower) {
 807                         initializeBehavior();
 808                     }
 809                 }
 810             });
 811         }
 812     }
 813
 814     /**
 815      * @deprecated Deprecated in favor of {@link org.opendaylight.controller.cluster.raft.base.messages.DeleteEntries}
 816      *             whose type for fromIndex is long instead of int. This class was kept for backwards
 817      *             compatibility with Helium.
 818      */
 819     // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility.
 820     @SuppressWarnings("serial")
 821     @Deprecated
 822     static class DeleteEntries implements Serializable {
 823         private final int fromIndex;
 824
 825         public DeleteEntries(int fromIndex) {
 826             this.fromIndex = fromIndex;
 827         }
 828
 829         public int getFromIndex() {
 830             return fromIndex;
 831         }
 832     }
 833
 834     /**
 835      * @deprecated Deprecated in favor of non-inner class {@link org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm}
 836      *             which has serialVersionUID set. This class was kept for backwards compatibility with Helium.
 837      */
 838     // Suppressing this warning as we can't set serialVersionUID to maintain backwards compatibility.
 839     @SuppressWarnings("serial")
 840     @Deprecated
 841     static class UpdateElectionTerm implements Serializable {
 842         private final long currentTerm;
 843         private final String votedFor;
 844
 845         public UpdateElectionTerm(long currentTerm, String votedFor) {
 846             this.currentTerm = currentTerm;
 847             this.votedFor = votedFor;
 848         }
 849
 850         public long getCurrentTerm() {
 851             return currentTerm;
 852         }
 853
 854         public String getVotedFor() {
 855             return votedFor;
 856         }
 857     }
 858
 859     private static class BehaviorStateHolder {
 860         private RaftActorBehavior behavior;
 861         private String lastValidLeaderId;
 862         private short leaderPayloadVersion;
 863
 864         void init(RaftActorBehavior behavior) {
 865             this.behavior = behavior;
 866             this.leaderPayloadVersion = behavior != null ? behavior.getLeaderPayloadVersion() : -1;
 867
 868             String behaviorLeaderId = behavior != null ? behavior.getLeaderId() : null;
 869             if(behaviorLeaderId != null) {
 870                 this.lastValidLeaderId = behaviorLeaderId;
 871             }
 872         }
 873
 874         RaftActorBehavior getBehavior() {
 875             return behavior;
 876         }
 877
 878         String getLastValidLeaderId() {
 879             return lastValidLeaderId;
 880         }
 881
 882         short getLeaderPayloadVersion() {
 883             return leaderPayloadVersion;
 884         }
 885     }
 886 }