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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.