Move ServerConfigurationPayload to cluster.raft.persisted
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / RaftActorServerConfigurationSupport.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.base.Preconditions;
14 import java.util.ArrayDeque;
15 import java.util.ArrayList;
16 import java.util.Collection;
17 import java.util.HashSet;
18 import java.util.List;
19 import java.util.Map;
20 import java.util.Queue;
21 import java.util.UUID;
22 import javax.annotation.Nullable;
23 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
24 import org.opendaylight.controller.cluster.raft.base.messages.SnapshotComplete;
25 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
26 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
27 import org.opendaylight.controller.cluster.raft.messages.AddServer;
28 import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
29 import org.opendaylight.controller.cluster.raft.messages.ChangeServersVotingStatus;
30 import org.opendaylight.controller.cluster.raft.messages.RemoveServer;
31 import org.opendaylight.controller.cluster.raft.messages.RemoveServerReply;
32 import org.opendaylight.controller.cluster.raft.messages.ServerChangeReply;
33 import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
34 import org.opendaylight.controller.cluster.raft.messages.ServerRemoved;
35 import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply;
36 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
37 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
38 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
39 import org.opendaylight.yangtools.concepts.Identifier;
40 import org.opendaylight.yangtools.util.AbstractUUIDIdentifier;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
43 import scala.concurrent.duration.FiniteDuration;
44
45 /**
46  * Handles server configuration related messages for a RaftActor.
47  *
48  * @author Thomas Pantelis
49  */
50 class RaftActorServerConfigurationSupport {
51     private static final Logger LOG = LoggerFactory.getLogger(RaftActorServerConfigurationSupport.class);
52
53     private final OperationState IDLE = new Idle();
54
55     private final RaftActor raftActor;
56
57     private final RaftActorContext raftContext;
58
59     private final Queue<ServerOperationContext<?>> pendingOperationsQueue = new ArrayDeque<>();
60
61     private OperationState currentOperationState = IDLE;
62
63     RaftActorServerConfigurationSupport(RaftActor raftActor) {
64         this.raftActor = raftActor;
65         this.raftContext = raftActor.getRaftActorContext();
66     }
67
68     boolean handleMessage(Object message, ActorRef sender) {
69         if(message instanceof AddServer) {
70             onAddServer((AddServer) message, sender);
71             return true;
72         } else if(message instanceof RemoveServer) {
73             onRemoveServer((RemoveServer) message, sender);
74             return true;
75         } else if(message instanceof ChangeServersVotingStatus) {
76             onChangeServersVotingStatus((ChangeServersVotingStatus) message, sender);
77             return true;
78         } else if (message instanceof ServerOperationTimeout) {
79             currentOperationState.onServerOperationTimeout((ServerOperationTimeout) message);
80             return true;
81         } else if (message instanceof UnInitializedFollowerSnapshotReply) {
82             currentOperationState.onUnInitializedFollowerSnapshotReply((UnInitializedFollowerSnapshotReply) message);
83             return true;
84         } else if(message instanceof ApplyState) {
85             return onApplyState((ApplyState) message);
86         } else if(message instanceof SnapshotComplete) {
87             currentOperationState.onSnapshotComplete();
88             return false;
89         } else {
90             return false;
91         }
92     }
93
94     void onNewLeader(String leaderId) {
95         currentOperationState.onNewLeader(leaderId);
96     }
97
98     private void onChangeServersVotingStatus(ChangeServersVotingStatus message, ActorRef sender) {
99         LOG.debug("{}: onChangeServersVotingStatus: {}, state: {}", raftContext.getId(), message,
100                 currentOperationState);
101
102         // The following check is a special case. Normally we fail an operation if there's no leader.
103         // Consider a scenario where one has 2 geographically-separated 3-node clusters, one a primary and
104         // the other a backup such that if the primary cluster is lost, the backup can take over. In this
105         // scenario, we have a logical 6-node cluster where the primary sub-cluster is configured as voting
106         // and the backup sub-cluster as non-voting such that the primary cluster can make progress without
107         // consensus from the backup cluster while still replicating to the backup. On fail-over to the backup,
108         // a request would be sent to a member of the backup cluster to flip the voting states, ie make the
109         // backup sub-cluster voting and the lost primary non-voting. However since the primary majority
110         // cluster is lost, there would be no leader to apply, persist and replicate the server config change.
111         // Therefore, if the local server is currently non-voting and is to be changed to voting and there is
112         // no current leader, we will try to elect a leader using the new server config in order to replicate
113         // the change and progress.
114         boolean localServerChangingToVoting = Boolean.TRUE.equals(message.
115                 getServerVotingStatusMap().get(raftActor.getRaftActorContext().getId()));
116         boolean hasNoLeader = raftActor.getLeaderId() == null;
117         if(localServerChangingToVoting && !raftContext.isVotingMember() && hasNoLeader) {
118             currentOperationState.onNewOperation(new ChangeServersVotingStatusContext(message, sender, true));
119         } else {
120             onNewOperation(new ChangeServersVotingStatusContext(message, sender, false));
121         }
122     }
123
124     private void onRemoveServer(RemoveServer removeServer, ActorRef sender) {
125         LOG.debug("{}: onRemoveServer: {}, state: {}", raftContext.getId(), removeServer, currentOperationState);
126         boolean isSelf = removeServer.getServerId().equals(raftContext.getId());
127         if(isSelf && !raftContext.hasFollowers()) {
128             sender.tell(new RemoveServerReply(ServerChangeStatus.NOT_SUPPORTED, raftActor.getLeaderId()),
129                     raftActor.getSelf());
130         } else if(!isSelf && !raftContext.getPeerIds().contains(removeServer.getServerId())) {
131             sender.tell(new RemoveServerReply(ServerChangeStatus.DOES_NOT_EXIST, raftActor.getLeaderId()),
132                     raftActor.getSelf());
133         } else {
134             String serverAddress = isSelf ? raftActor.self().path().toString() :
135                 raftContext.getPeerAddress(removeServer.getServerId());
136             onNewOperation(new RemoveServerContext(removeServer, serverAddress, sender));
137         }
138     }
139
140     private boolean onApplyState(ApplyState applyState) {
141         Payload data = applyState.getReplicatedLogEntry().getData();
142         if(data instanceof ServerConfigurationPayload) {
143             currentOperationState.onApplyState(applyState);
144             return true;
145         }
146
147         return false;
148     }
149
150     /**
151      * The algorithm for AddServer is as follows:
152      * <ul>
153      * <li>Add the new server as a peer.</li>
154      * <li>Add the new follower to the leader.</li>
155      * <li>If new server should be voting member</li>
156      * <ul>
157      *     <li>Initialize FollowerState to VOTING_NOT_INITIALIZED.</li>
158      *     <li>Initiate install snapshot to the new follower.</li>
159      *     <li>When install snapshot complete, mark the follower as VOTING and re-calculate majority vote count.</li>
160      * </ul>
161      * <li>Persist and replicate ServerConfigurationPayload with the new server list.</li>
162      * <li>On replication consensus, respond to caller with OK.</li>
163      * </ul>
164      * If the install snapshot times out after a period of 2 * election time out
165      * <ul>
166      *     <li>Remove the new server as a peer.</li>
167      *     <li>Remove the new follower from the leader.</li>
168      *     <li>Respond to caller with TIMEOUT.</li>
169      * </ul>
170      */
171     private void onAddServer(AddServer addServer, ActorRef sender) {
172         LOG.debug("{}: onAddServer: {}, state: {}", raftContext.getId(), addServer, currentOperationState);
173
174         onNewOperation(new AddServerContext(addServer, sender));
175     }
176
177     private void onNewOperation(ServerOperationContext<?> operationContext) {
178         if (raftActor.isLeader()) {
179             currentOperationState.onNewOperation(operationContext);
180         } else {
181             ActorSelection leader = raftActor.getLeader();
182             if (leader != null) {
183                 LOG.debug("{}: Not leader - forwarding to leader {}", raftContext.getId(), leader);
184                 leader.tell(operationContext.getOperation(), operationContext.getClientRequestor());
185             } else {
186                 LOG.debug("{}: No leader - returning NO_LEADER reply", raftContext.getId());
187                 operationContext.getClientRequestor().tell(operationContext.newReply(
188                         ServerChangeStatus.NO_LEADER, null), raftActor.self());
189             }
190         }
191     }
192
193     /**
194      * Interface for the initial state for a server operation.
195      */
196     private interface InitialOperationState {
197         void initiate();
198     }
199
200     /**
201      * Abstract base class for a server operation FSM state. Handles common behavior for all states.
202      */
203     private abstract class OperationState {
204         void onNewOperation(ServerOperationContext<?> operationContext) {
205             // We're currently processing another operation so queue it to be processed later.
206
207             LOG.debug("{}: Server operation already in progress - queueing {}", raftContext.getId(),
208                     operationContext.getOperation());
209
210             pendingOperationsQueue.add(operationContext);
211         }
212
213         void onServerOperationTimeout(ServerOperationTimeout timeout) {
214             LOG.debug("onServerOperationTimeout should not be called in state {}", this);
215         }
216
217         void onUnInitializedFollowerSnapshotReply(UnInitializedFollowerSnapshotReply reply) {
218             LOG.debug("onUnInitializedFollowerSnapshotReply was called in state {}", this);
219         }
220
221         void onApplyState(ApplyState applyState) {
222             LOG.debug("onApplyState was called in state {}", this);
223         }
224
225         void onSnapshotComplete() {
226
227         }
228
229         void onNewLeader(String newLeader) {
230         }
231
232         protected void persistNewServerConfiguration(ServerOperationContext<?> operationContext){
233             raftContext.setDynamicServerConfigurationInUse();
234
235             ServerConfigurationPayload payload = raftContext.getPeerServerInfo(
236                     operationContext.includeSelfInNewConfiguration(raftActor));
237             LOG.debug("{}: New server configuration : {}", raftContext.getId(), payload.getServerConfig());
238
239             raftActor.persistData(operationContext.getClientRequestor(), operationContext.getContextId(), payload);
240
241             currentOperationState = new Persisting(operationContext, newTimer(new ServerOperationTimeout(
242                     operationContext.getLoggingContext())));
243
244             sendReply(operationContext, ServerChangeStatus.OK);
245         }
246
247         protected void operationComplete(ServerOperationContext<?> operationContext, @Nullable ServerChangeStatus replyStatus) {
248             if(replyStatus != null) {
249                 sendReply(operationContext, replyStatus);
250             }
251
252             operationContext.operationComplete(raftActor, replyStatus == null || replyStatus == ServerChangeStatus.OK);
253
254             changeToIdleState();
255         }
256
257         protected void changeToIdleState() {
258             currentOperationState = IDLE;
259
260             ServerOperationContext<?> nextOperation = pendingOperationsQueue.poll();
261             if(nextOperation != null) {
262                 RaftActorServerConfigurationSupport.this.onNewOperation(nextOperation);
263             }
264         }
265
266         protected void sendReply(ServerOperationContext<?> operationContext, ServerChangeStatus status) {
267             LOG.debug("{}: Returning {} for operation {}", raftContext.getId(), status, operationContext.getOperation());
268
269             operationContext.getClientRequestor().tell(operationContext.newReply(status, raftActor.getLeaderId()),
270                     raftActor.self());
271         }
272
273         Cancellable newTimer(Object message) {
274             return newTimer(raftContext.getConfigParams().getElectionTimeOutInterval().$times(2), message);
275         }
276
277         Cancellable newTimer(FiniteDuration timeout, Object message) {
278             return raftContext.getActorSystem().scheduler().scheduleOnce(
279                     timeout, raftContext.getActor(), message,
280                             raftContext.getActorSystem().dispatcher(), raftContext.getActor());
281         }
282
283         @Override
284         public String toString() {
285             return getClass().getSimpleName();
286         }
287     }
288
289     /**
290      * The state when no server operation is in progress. It immediately initiates new server operations.
291      */
292     private final class Idle extends OperationState {
293         @Override
294         public void onNewOperation(ServerOperationContext<?> operationContext) {
295             operationContext.newInitialOperationState(RaftActorServerConfigurationSupport.this).initiate();
296         }
297
298         @Override
299         public void onApplyState(ApplyState applyState) {
300             // Noop - we override b/c ApplyState is called normally for followers in the idle state.
301         }
302     }
303
304     /**
305      * The state when a new server configuration is being persisted and replicated.
306      */
307     private final class Persisting extends OperationState {
308         private final ServerOperationContext<?> operationContext;
309         private final Cancellable timer;
310         private boolean timedOut = false;
311
312         Persisting(ServerOperationContext<?> operationContext, Cancellable timer) {
313             this.operationContext = operationContext;
314             this.timer = timer;
315         }
316
317         @Override
318         public void onApplyState(ApplyState applyState) {
319             // Sanity check - we could get an ApplyState from a previous operation that timed out so make
320             // sure it's meant for us.
321             if(operationContext.getContextId().equals(applyState.getIdentifier())) {
322                 LOG.info("{}: {} has been successfully replicated to a majority of followers", raftContext.getId(),
323                         applyState.getReplicatedLogEntry().getData());
324
325                 timer.cancel();
326                 operationComplete(operationContext, null);
327             }
328         }
329
330         @Override
331         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
332             LOG.warn("{}: Timeout occured while replicating the new server configuration for {}", raftContext.getId(),
333                     timeout.getLoggingContext());
334
335             timedOut = true;
336
337             // Fail any pending operations
338             ServerOperationContext<?> nextOperation = pendingOperationsQueue.poll();
339             while(nextOperation != null) {
340                 sendReply(nextOperation, ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT);
341                 nextOperation = pendingOperationsQueue.poll();
342             }
343         }
344
345         @Override
346         public void onNewOperation(ServerOperationContext<?> operationContext) {
347             if(timedOut) {
348                 sendReply(operationContext, ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT);
349             } else {
350                 super.onNewOperation(operationContext);
351             }
352         }
353     }
354
355     /**
356      * Abstract base class for an AddServer operation state.
357      */
358     private abstract class AddServerState extends OperationState {
359         private final AddServerContext addServerContext;
360
361         AddServerState(AddServerContext addServerContext) {
362             this.addServerContext = addServerContext;
363         }
364
365         AddServerContext getAddServerContext() {
366             return addServerContext;
367         }
368
369         Cancellable newInstallSnapshotTimer() {
370             return newTimer(new ServerOperationTimeout(addServerContext.getOperation().getNewServerId()));
371         }
372
373         void handleInstallSnapshotTimeout(ServerOperationTimeout timeout) {
374             String serverId = timeout.getLoggingContext();
375
376             LOG.debug("{}: handleInstallSnapshotTimeout for new server {}", raftContext.getId(), serverId);
377
378             // cleanup
379             raftContext.removePeer(serverId);
380
381             boolean isLeader = raftActor.isLeader();
382             if(isLeader) {
383                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
384                 leader.removeFollower(serverId);
385             }
386
387             operationComplete(getAddServerContext(), isLeader ? ServerChangeStatus.TIMEOUT : ServerChangeStatus.NO_LEADER);
388         }
389
390     }
391
392     /**
393      * The initial state for the AddServer operation. It adds the new follower as a peer and initiates
394      * snapshot capture, if necessary.
395      */
396     private final class InitialAddServerState extends AddServerState implements InitialOperationState {
397         InitialAddServerState(AddServerContext addServerContext) {
398             super(addServerContext);
399         }
400
401         @Override
402         public void initiate() {
403             AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
404             AddServer addServer = getAddServerContext().getOperation();
405
406             LOG.debug("{}: Initiating {}", raftContext.getId(), addServer);
407
408             if(raftContext.getPeerInfo(addServer.getNewServerId()) != null) {
409                 operationComplete(getAddServerContext(), ServerChangeStatus.ALREADY_EXISTS);
410                 return;
411             }
412
413             VotingState votingState = addServer.isVotingMember() ? VotingState.VOTING_NOT_INITIALIZED :
414                     VotingState.NON_VOTING;
415             raftContext.addToPeers(addServer.getNewServerId(), addServer.getNewServerAddress(), votingState);
416
417             leader.addFollower(addServer.getNewServerId());
418
419             if(votingState == VotingState.VOTING_NOT_INITIALIZED){
420                 // schedule the install snapshot timeout timer
421                 Cancellable installSnapshotTimer = newInstallSnapshotTimer();
422                 if(leader.initiateCaptureSnapshot(addServer.getNewServerId())) {
423                     LOG.debug("{}: Initiating capture snapshot for new server {}", raftContext.getId(),
424                             addServer.getNewServerId());
425
426                     currentOperationState = new InstallingSnapshot(getAddServerContext(), installSnapshotTimer);
427                 } else {
428                     LOG.debug("{}: Snapshot already in progress - waiting for completion", raftContext.getId());
429
430                     currentOperationState = new WaitingForPriorSnapshotComplete(getAddServerContext(),
431                             installSnapshotTimer);
432                 }
433             } else {
434                 LOG.debug("{}: New follower is non-voting - directly persisting new server configuration",
435                         raftContext.getId());
436
437                 persistNewServerConfiguration(getAddServerContext());
438             }
439         }
440     }
441
442     /**
443      * The AddServer operation state for when the catch-up snapshot is being installed. It handles successful
444      * reply or timeout.
445      */
446     private final class InstallingSnapshot extends AddServerState {
447         private final Cancellable installSnapshotTimer;
448
449         InstallingSnapshot(AddServerContext addServerContext, Cancellable installSnapshotTimer) {
450             super(addServerContext);
451             this.installSnapshotTimer = Preconditions.checkNotNull(installSnapshotTimer);
452         }
453
454         @Override
455         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
456             handleInstallSnapshotTimeout(timeout);
457
458             LOG.warn("{}: Timeout occured for new server {} while installing snapshot", raftContext.getId(),
459                     timeout.getLoggingContext());
460         }
461
462         @Override
463         public void onUnInitializedFollowerSnapshotReply(UnInitializedFollowerSnapshotReply reply) {
464             LOG.debug("{}: onUnInitializedFollowerSnapshotReply: {}", raftContext.getId(), reply);
465
466             String followerId = reply.getFollowerId();
467
468             // Sanity check to guard against receiving an UnInitializedFollowerSnapshotReply from a prior
469             // add server operation that timed out.
470             if(getAddServerContext().getOperation().getNewServerId().equals(followerId) && raftActor.isLeader()) {
471                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
472                 raftContext.getPeerInfo(followerId).setVotingState(VotingState.VOTING);
473                 leader.updateMinReplicaCount();
474
475                 persistNewServerConfiguration(getAddServerContext());
476
477                 installSnapshotTimer.cancel();
478             } else {
479                 LOG.debug("{}: Dropping UnInitializedFollowerSnapshotReply for server {}: {}",
480                         raftContext.getId(), followerId,
481                         !raftActor.isLeader() ? "not leader" : "server Id doesn't match");
482             }
483         }
484     }
485
486     /**
487      * The AddServer operation state for when there is a snapshot already in progress. When the current
488      * snapshot completes, it initiates an install snapshot.
489      */
490     private final class WaitingForPriorSnapshotComplete extends AddServerState {
491         private final Cancellable snapshotTimer;
492
493         WaitingForPriorSnapshotComplete(AddServerContext addServerContext, Cancellable snapshotTimer) {
494             super(addServerContext);
495             this.snapshotTimer = Preconditions.checkNotNull(snapshotTimer);
496         }
497
498         @Override
499         public void onSnapshotComplete() {
500             LOG.debug("{}: onSnapshotComplete", raftContext.getId());
501
502             if(!raftActor.isLeader()) {
503                 LOG.debug("{}: No longer the leader", raftContext.getId());
504                 return;
505             }
506
507             AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
508             if(leader.initiateCaptureSnapshot(getAddServerContext().getOperation().getNewServerId())) {
509                 LOG.debug("{}: Initiating capture snapshot for new server {}", raftContext.getId(),
510                         getAddServerContext().getOperation().getNewServerId());
511
512                 currentOperationState = new InstallingSnapshot(getAddServerContext(),
513                         newInstallSnapshotTimer());
514
515                 snapshotTimer.cancel();
516             }
517         }
518
519         @Override
520         public void onServerOperationTimeout(ServerOperationTimeout timeout) {
521             handleInstallSnapshotTimeout(timeout);
522
523             LOG.warn("{}: Timeout occured for new server {} while waiting for prior snapshot to complete",
524                     raftContext.getId(), timeout.getLoggingContext());
525         }
526     }
527
528     private static final class ServerOperationContextIdentifier extends AbstractUUIDIdentifier<ServerOperationContextIdentifier> {
529         private static final long serialVersionUID = 1L;
530
531         ServerOperationContextIdentifier() {
532             super(UUID.randomUUID());
533         }
534     }
535
536     /**
537      * Stores context information for a server operation.
538      *
539      * @param <T> the operation type
540      */
541     private static abstract class ServerOperationContext<T> {
542         private final T operation;
543         private final ActorRef clientRequestor;
544         private final Identifier contextId;
545
546         ServerOperationContext(T operation, ActorRef clientRequestor){
547             this.operation = operation;
548             this.clientRequestor = clientRequestor;
549             contextId = new ServerOperationContextIdentifier();
550         }
551
552         Identifier getContextId() {
553             return contextId;
554         }
555
556         T getOperation() {
557             return operation;
558         }
559
560         ActorRef getClientRequestor() {
561             return clientRequestor;
562         }
563
564         void operationComplete(RaftActor raftActor, boolean succeeded) {
565         }
566
567         boolean includeSelfInNewConfiguration(RaftActor raftActor) {
568             return true;
569         }
570
571         abstract Object newReply(ServerChangeStatus status, String leaderId);
572
573         abstract InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support);
574
575         abstract String getLoggingContext();
576     }
577
578     /**
579      * Stores context information for an AddServer operation.
580      */
581     private static class AddServerContext extends ServerOperationContext<AddServer> {
582         AddServerContext(AddServer addServer, ActorRef clientRequestor) {
583             super(addServer, clientRequestor);
584         }
585
586         @Override
587         Object newReply(ServerChangeStatus status, String leaderId) {
588             return new AddServerReply(status, leaderId);
589         }
590
591         @Override
592         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
593             return support.new InitialAddServerState(this);
594         }
595
596         @Override
597         String getLoggingContext() {
598             return getOperation().getNewServerId();
599         }
600     }
601
602     private abstract class RemoveServerState extends OperationState {
603         private final RemoveServerContext removeServerContext;
604
605         protected RemoveServerState(RemoveServerContext removeServerContext) {
606             this.removeServerContext = Preconditions.checkNotNull(removeServerContext);
607
608         }
609
610         public RemoveServerContext getRemoveServerContext() {
611             return removeServerContext;
612         }
613     }
614
615     private final class InitialRemoveServerState extends RemoveServerState implements InitialOperationState{
616
617         protected InitialRemoveServerState(RemoveServerContext removeServerContext) {
618             super(removeServerContext);
619         }
620
621         @Override
622         public void initiate() {
623             String serverId = getRemoveServerContext().getOperation().getServerId();
624             raftContext.removePeer(serverId);
625             ((AbstractLeader)raftActor.getCurrentBehavior()).removeFollower(serverId);
626
627             persistNewServerConfiguration(getRemoveServerContext());
628         }
629     }
630
631     private static class RemoveServerContext extends ServerOperationContext<RemoveServer> {
632         private final String peerAddress;
633
634         RemoveServerContext(RemoveServer operation, String peerAddress, ActorRef clientRequestor) {
635             super(operation, clientRequestor);
636             this.peerAddress = peerAddress;
637         }
638
639         @Override
640         Object newReply(ServerChangeStatus status, String leaderId) {
641             return new RemoveServerReply(status, leaderId);
642         }
643
644         @Override
645         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
646             return support.new InitialRemoveServerState(this);
647         }
648
649         @Override
650         void operationComplete(RaftActor raftActor, boolean succeeded) {
651             if(peerAddress != null) {
652                 raftActor.context().actorSelection(peerAddress).tell(new ServerRemoved(getOperation().getServerId()), raftActor.getSelf());
653             }
654         }
655
656         @Override
657         boolean includeSelfInNewConfiguration(RaftActor raftActor) {
658             return !getOperation().getServerId().equals(raftActor.getId());
659         }
660
661         @Override
662         String getLoggingContext() {
663             return getOperation().getServerId();
664         }
665     }
666
667     private static class ChangeServersVotingStatusContext extends ServerOperationContext<ChangeServersVotingStatus> {
668         private final boolean tryToElectLeader;
669
670         ChangeServersVotingStatusContext(ChangeServersVotingStatus convertMessage, ActorRef clientRequestor,
671                 boolean tryToElectLeader) {
672             super(convertMessage, clientRequestor);
673             this.tryToElectLeader = tryToElectLeader;
674         }
675
676         @Override
677         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
678             return support.new ChangeServersVotingStatusState(this, tryToElectLeader);
679         }
680
681         @Override
682         Object newReply(ServerChangeStatus status, String leaderId) {
683             return new ServerChangeReply(status, leaderId);
684         }
685
686         @Override
687         void operationComplete(final RaftActor raftActor, boolean succeeded) {
688             // If this leader changed to non-voting we need to step down as leader so we'll try to transfer
689             // leadership.
690             boolean localServerChangedToNonVoting = Boolean.FALSE.equals(getOperation().
691                     getServerVotingStatusMap().get(raftActor.getRaftActorContext().getId()));
692             if (succeeded && localServerChangedToNonVoting) {
693                 LOG.debug("Leader changed to non-voting - trying leadership transfer");
694                 raftActor.becomeNonVoting();
695             }
696         }
697
698         @Override
699         String getLoggingContext() {
700             return getOperation().toString();
701         }
702     }
703
704     private class ChangeServersVotingStatusState extends OperationState implements InitialOperationState {
705         private final ChangeServersVotingStatusContext changeVotingStatusContext;
706         private final boolean tryToElectLeader;
707
708         ChangeServersVotingStatusState(ChangeServersVotingStatusContext changeVotingStatusContext,
709                 boolean tryToElectLeader) {
710             this.changeVotingStatusContext = changeVotingStatusContext;
711             this.tryToElectLeader = tryToElectLeader;
712         }
713
714         @Override
715         public void initiate() {
716             LOG.debug("Initiating ChangeServersVotingStatusState");
717
718             if(tryToElectLeader) {
719                 initiateLocalLeaderElection();
720             } else if(updateLocalPeerInfo()) {
721                 persistNewServerConfiguration(changeVotingStatusContext);
722             }
723         }
724
725         private void initiateLocalLeaderElection() {
726             LOG.debug("{}: Sending local ElectionTimeout to start leader election", raftContext.getId());
727
728             ServerConfigurationPayload previousServerConfig = raftContext.getPeerServerInfo(true);
729             if(!updateLocalPeerInfo()) {
730                 return;
731             }
732
733             raftContext.getActor().tell(TimeoutNow.INSTANCE, raftContext.getActor());
734
735             currentOperationState = new WaitingForLeaderElected(changeVotingStatusContext, previousServerConfig);
736         }
737
738         private boolean updateLocalPeerInfo() {
739             List<ServerInfo> newServerInfoList = newServerInfoList();
740
741             // Check if new voting state would leave us with no voting members.
742             boolean atLeastOneVoting = false;
743             for(ServerInfo info: newServerInfoList) {
744                 if(info.isVoting()) {
745                     atLeastOneVoting = true;
746                     break;
747                 }
748             }
749
750             if(!atLeastOneVoting) {
751                 operationComplete(changeVotingStatusContext, ServerChangeStatus.INVALID_REQUEST);
752                 return false;
753             }
754
755             raftContext.updatePeerIds(new ServerConfigurationPayload(newServerInfoList));
756             if(raftActor.getCurrentBehavior() instanceof AbstractLeader) {
757                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
758                 leader.updateMinReplicaCount();
759             }
760
761             return true;
762         }
763
764         private List<ServerInfo> newServerInfoList() {
765             Map<String, Boolean> serverVotingStatusMap = changeVotingStatusContext.getOperation().getServerVotingStatusMap();
766             List<ServerInfo> newServerInfoList = new ArrayList<>();
767             for(String peerId: raftContext.getPeerIds()) {
768                 newServerInfoList.add(new ServerInfo(peerId, serverVotingStatusMap.containsKey(peerId) ?
769                         serverVotingStatusMap.get(peerId) : raftContext.getPeerInfo(peerId).isVoting()));
770             }
771
772             newServerInfoList.add(new ServerInfo(raftContext.getId(), serverVotingStatusMap.containsKey(
773                     raftContext.getId()) ? serverVotingStatusMap.get(raftContext.getId()) : raftContext.isVotingMember()));
774
775             return newServerInfoList;
776         }
777     }
778
779     private class WaitingForLeaderElected extends OperationState {
780         private final ServerConfigurationPayload previousServerConfig;
781         private final ChangeServersVotingStatusContext operationContext;
782         private final Cancellable timer;
783
784         WaitingForLeaderElected(ChangeServersVotingStatusContext operationContext,
785                 ServerConfigurationPayload previousServerConfig) {
786             this.operationContext = operationContext;
787             this.previousServerConfig = previousServerConfig;
788
789             timer = newTimer(raftContext.getConfigParams().getElectionTimeOutInterval(),
790                     new ServerOperationTimeout(operationContext.getLoggingContext()));
791         }
792
793         @Override
794         void onNewLeader(String newLeader) {
795             if(newLeader == null) {
796                 return;
797             }
798
799             LOG.debug("{}: New leader {} elected", raftContext.getId(), newLeader);
800
801             timer.cancel();
802
803             if(raftActor.isLeader()) {
804                 persistNewServerConfiguration(operationContext);
805             } else {
806                 // Edge case - some other node became leader so forward the operation.
807                 LOG.debug("{}: Forwarding {} to new leader", raftContext.getId(), operationContext.getOperation());
808
809                 // Revert the local server config change.
810                 raftContext.updatePeerIds(previousServerConfig);
811
812                 changeToIdleState();
813                 RaftActorServerConfigurationSupport.this.onNewOperation(operationContext);
814             }
815         }
816
817         @Override
818         void onServerOperationTimeout(ServerOperationTimeout timeout) {
819             LOG.warn("{}: Leader election timed out - cannot apply operation {}",
820                     raftContext.getId(), timeout.getLoggingContext());
821
822             // Revert the local server config change.
823             raftContext.updatePeerIds(previousServerConfig);
824             raftActor.initializeBehavior();
825
826             tryToForwardOperationToAnotherServer();
827         }
828
829         private void tryToForwardOperationToAnotherServer() {
830             Collection<String> serversVisited = new HashSet<>(operationContext.getOperation().getServersVisited());
831
832             LOG.debug("{}: tryToForwardOperationToAnotherServer - servers already visited {}", raftContext.getId(),
833                     serversVisited);
834
835             serversVisited.add(raftContext.getId());
836
837             // Try to find another whose state is being changed from non-voting to voting and that we haven't
838             // tried yet.
839             Map<String, Boolean> serverVotingStatusMap = operationContext.getOperation().getServerVotingStatusMap();
840             ActorSelection forwardToPeerActor = null;
841             for(Map.Entry<String, Boolean> e: serverVotingStatusMap.entrySet()) {
842                 Boolean isVoting = e.getValue();
843                 String serverId = e.getKey();
844                 PeerInfo peerInfo = raftContext.getPeerInfo(serverId);
845                 if(isVoting && peerInfo != null && !peerInfo.isVoting() && !serversVisited.contains(serverId)) {
846                     ActorSelection actor = raftContext.getPeerActorSelection(serverId);
847                     if(actor != null) {
848                         forwardToPeerActor = actor;
849                         break;
850                     }
851                 }
852             }
853
854             if(forwardToPeerActor != null) {
855                 LOG.debug("{}: Found server {} to forward to", raftContext.getId(), forwardToPeerActor);
856
857                 forwardToPeerActor.tell(new ChangeServersVotingStatus(serverVotingStatusMap, serversVisited),
858                         operationContext.getClientRequestor());
859                 changeToIdleState();
860             } else {
861                 operationComplete(operationContext, ServerChangeStatus.NO_LEADER);
862             }
863         }
864     }
865
866     static class ServerOperationTimeout {
867         private final String loggingContext;
868
869         ServerOperationTimeout(String loggingContext){
870            this.loggingContext = Preconditions.checkNotNull(loggingContext, "loggingContext should not be null");
871         }
872
873         String getLoggingContext() {
874             return loggingContext;
875         }
876     }
877 }

©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.