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