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.ArrayList;
  15 import java.util.Collections;
  16 import java.util.LinkedList;
  17 import java.util.List;
  18 import java.util.Queue;
  19 import java.util.UUID;
  20 import java.util.concurrent.TimeUnit;
  21 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  22 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
  23 import org.opendaylight.controller.cluster.raft.messages.AddServer;
  24 import org.opendaylight.controller.cluster.raft.messages.AddServerReply;
  25 import org.opendaylight.controller.cluster.raft.messages.FollowerCatchUpTimeout;
  26 import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus;
  27 import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply;
  28 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  29 import org.slf4j.Logger;
  30 import org.slf4j.LoggerFactory;
  31 import scala.concurrent.duration.FiniteDuration;
  32
  33 /**
  34  * Handles server configuration related messages for a RaftActor.
  35  *
  36  * @author Thomas Pantelis
  37  */
  38 class RaftActorServerConfigurationSupport {
  39     private static final Logger LOG = LoggerFactory.getLogger(RaftActorServerConfigurationSupport.class);
  40
  41     private final OperationState IDLE = new Idle();
  42
  43     private final RaftActorContext raftContext;
  44
  45     private final Queue<ServerOperationContext<?>> pendingOperationsQueue = new LinkedList<>();
  46
  47     private OperationState currentOperationState = IDLE;
  48
  49     RaftActorServerConfigurationSupport(RaftActorContext context) {
  50         this.raftContext = context;
  51     }
  52
  53     boolean handleMessage(Object message, RaftActor raftActor, ActorRef sender) {
  54         if(message instanceof AddServer) {
  55             onAddServer((AddServer)message, raftActor, sender);
  56             return true;
  57         } else if (message instanceof FollowerCatchUpTimeout) {
  58             currentOperationState.onFollowerCatchupTimeout(raftActor, (FollowerCatchUpTimeout)message);
  59             return true;
  60         } else if (message instanceof UnInitializedFollowerSnapshotReply) {
  61             currentOperationState.onUnInitializedFollowerSnapshotReply(raftActor,
  62                     (UnInitializedFollowerSnapshotReply)message);
  63             return true;
  64         } else if(message instanceof ApplyState) {
  65             return onApplyState((ApplyState) message, raftActor);
  66         } else {
  67             return false;
  68         }
  69     }
  70
  71     private boolean onApplyState(ApplyState applyState, RaftActor raftActor) {
  72         Payload data = applyState.getReplicatedLogEntry().getData();
  73         if(data instanceof ServerConfigurationPayload) {
  74             currentOperationState.onApplyState(raftActor, applyState);
  75             return true;
  76         }
  77
  78         return false;
  79     }
  80
  81     /**
  82      * The algorithm for AddServer is as follows:
  83      * <ul>
  84      * <li>Add the new server as a peer.</li>
  85      * <li>Add the new follower to the leader.</li>
  86      * <li>If new server should be voting member</li>
  87      * <ul>
  88      *     <li>Initialize FollowerState to VOTING_NOT_INITIALIZED.</li>
  89      *     <li>Initiate install snapshot to the new follower.</li>
  90      *     <li>When install snapshot complete, mark the follower as VOTING and re-calculate majority vote count.</li>
  91      * </ul>
  92      * <li>Persist and replicate ServerConfigurationPayload with the new server list.</li>
  93      * <li>On replication consensus, respond to caller with OK.</li>
  94      * </ul>
  95      * If the install snapshot times out after a period of 2 * election time out
  96      * <ul>
  97      *     <li>Remove the new server as a peer.</li>
  98      *     <li>Remove the new follower from the leader.</li>
  99      *     <li>Respond to caller with TIMEOUT.</li>
 100      * </ul>
 101      */
 102     private void onAddServer(AddServer addServer, RaftActor raftActor, ActorRef sender) {
 103         LOG.debug("{}: onAddServer: {}", raftContext.getId(), addServer);
 104
 105         onNewOperation(raftActor, new AddServerContext(addServer, sender));
 106     }
 107
 108     private void onNewOperation(RaftActor raftActor, ServerOperationContext<?> operationContext) {
 109         if (raftActor.isLeader()) {
 110             currentOperationState.onNewOperation(raftActor, operationContext);
 111         } else {
 112             ActorSelection leader = raftActor.getLeader();
 113             if (leader != null) {
 114                 LOG.debug("{}: Not leader - forwarding to leader {}", raftContext.getId(), leader);
 115                 leader.forward(operationContext.getOperation(), raftActor.getContext());
 116             } else {
 117                 LOG.debug("{}: No leader - returning NO_LEADER reply", raftContext.getId());
 118                 operationContext.getClientRequestor().tell(operationContext.newReply(
 119                         ServerChangeStatus.NO_LEADER, null), raftActor.self());
 120             }
 121         }
 122     }
 123
 124     /**
 125      * Interface for a server operation FSM state.
 126      */
 127     private interface OperationState {
 128         void onNewOperation(RaftActor raftActor, ServerOperationContext<?> operationContext);
 129
 130         void onFollowerCatchupTimeout(RaftActor raftActor, FollowerCatchUpTimeout followerTimeout);
 131
 132         void onUnInitializedFollowerSnapshotReply(RaftActor raftActor, UnInitializedFollowerSnapshotReply reply);
 133
 134         void onApplyState(RaftActor raftActor, ApplyState applyState);
 135     }
 136
 137     /**
 138      * Interface for the initial state for a server operation.
 139      */
 140     private interface InitialOperationState {
 141         void initiate(RaftActor raftActor);
 142     }
 143
 144     /**
 145      * Abstract base class for server operation FSM state. Handles common behavior for all states.
 146      */
 147     private abstract class AbstractOperationState implements OperationState {
 148         @Override
 149         public void onNewOperation(RaftActor raftActor, ServerOperationContext<?> operationContext) {
 150             // We're currently processing another operation so queue it to be processed later.
 151
 152             LOG.debug("{}: Server operation already in progress - queueing {}", raftContext.getId(),
 153                     operationContext.getOperation());
 154
 155             pendingOperationsQueue.add(operationContext);
 156         }
 157
 158         @Override
 159         public void onFollowerCatchupTimeout(RaftActor raftActor, FollowerCatchUpTimeout followerTimeout) {
 160             LOG.debug("onFollowerCatchupTimeout should not be called in state {}", this);
 161         }
 162
 163         @Override
 164         public void onUnInitializedFollowerSnapshotReply(RaftActor raftActor, UnInitializedFollowerSnapshotReply reply) {
 165             LOG.debug("onUnInitializedFollowerSnapshotReply was called in state {}", this);
 166         }
 167
 168         @Override
 169         public void onApplyState(RaftActor raftActor, ApplyState applyState) {
 170             LOG.debug("onApplyState was called in state {}", this);
 171         }
 172
 173         protected void persistNewServerConfiguration(RaftActor raftActor, ServerOperationContext<?> operationContext){
 174             List <String> newConfig = new ArrayList<String>(raftContext.getPeerIds());
 175             newConfig.add(raftContext.getId());
 176
 177             LOG.debug("{}: New server configuration : {}", raftContext.getId(), newConfig);
 178
 179             ServerConfigurationPayload payload = new ServerConfigurationPayload(newConfig, Collections.<String>emptyList());
 180
 181             raftActor.persistData(operationContext.getClientRequestor(), operationContext.getContextId(), payload);
 182
 183             currentOperationState = new Persisting(operationContext);
 184         }
 185
 186         protected void operationComplete(RaftActor raftActor, ServerOperationContext<?> operationContext,
 187                 ServerChangeStatus status) {
 188
 189             LOG.debug("{}: Returning {} for operation {}", raftContext.getId(), status, operationContext.getOperation());
 190
 191             operationContext.getClientRequestor().tell(operationContext.newReply(status, raftActor.getLeaderId()),
 192                     raftActor.self());
 193
 194             currentOperationState = IDLE;
 195
 196             ServerOperationContext<?> nextOperation = pendingOperationsQueue.poll();
 197             if(nextOperation != null) {
 198                 RaftActorServerConfigurationSupport.this.onNewOperation(raftActor, nextOperation);
 199             }
 200         }
 201
 202         @Override
 203         public String toString() {
 204             return getClass().getSimpleName();
 205         }
 206     }
 207
 208     /**
 209      * The state when no server operation is in progress. It immediately initiates new server operations.
 210      */
 211     private class Idle extends AbstractOperationState {
 212         @Override
 213         public void onNewOperation(RaftActor raftActor, ServerOperationContext<?> operationContext) {
 214             operationContext.newInitialOperationState(RaftActorServerConfigurationSupport.this).initiate(raftActor);
 215         }
 216
 217         @Override
 218         public void onApplyState(RaftActor raftActor, ApplyState applyState) {
 219             // Noop - we override b/c ApplyState is called normally for followers in the idle state.
 220         }
 221     }
 222
 223     /**
 224      * The state when a new server configuration is being persisted and replicated.
 225      */
 226     private class Persisting extends AbstractOperationState {
 227         private final ServerOperationContext<?> operationContext;
 228
 229         Persisting(ServerOperationContext<?> operationContext) {
 230             this.operationContext = operationContext;
 231         }
 232
 233         @Override
 234         public void onApplyState(RaftActor raftActor, ApplyState applyState) {
 235             // Sanity check - we could get an ApplyState from a previous operation that timed out so make
 236             // sure it's meant for us.
 237             if(operationContext.getContextId().equals(applyState.getIdentifier())) {
 238                 LOG.info("{}: {} has been successfully replicated to a majority of followers",
 239                         applyState.getReplicatedLogEntry().getData());
 240
 241                 operationComplete(raftActor, operationContext, ServerChangeStatus.OK);
 242             }
 243         }
 244     }
 245
 246     /**
 247      * Abstract base class for an AddServer operation state.
 248      */
 249     private abstract class AddServerState extends AbstractOperationState {
 250         private final AddServerContext addServerContext;
 251
 252         AddServerState(AddServerContext addServerContext) {
 253             this.addServerContext = addServerContext;
 254         }
 255
 256         AddServerContext getAddServerContext() {
 257             return addServerContext;
 258         }
 259     }
 260
 261     /**
 262      * The initial state for the AddServer operation. It adds the new follower as a peer and initiates
 263      * snapshot capture, if necessary.
 264      */
 265     private class InitialAddServerState extends AddServerState implements InitialOperationState {
 266         InitialAddServerState(AddServerContext addServerContext) {
 267             super(addServerContext);
 268         }
 269
 270         @Override
 271         public void initiate(RaftActor raftActor) {
 272             AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
 273
 274             AddServer addServer = getAddServerContext().getOperation();
 275
 276             LOG.debug("{}: Initiating {}", raftContext.getId(), addServer);
 277
 278             VotingState votingState = addServer.isVotingMember() ? VotingState.VOTING_NOT_INITIALIZED :
 279                     VotingState.NON_VOTING;
 280             raftContext.addToPeers(addServer.getNewServerId(), addServer.getNewServerAddress(), votingState);
 281
 282             leader.addFollower(addServer.getNewServerId());
 283
 284             if(votingState == VotingState.VOTING_NOT_INITIALIZED){
 285                 LOG.debug("{}: Leader sending initiate capture snapshot to new follower {}", raftContext.getId(),
 286                         addServer.getNewServerId());
 287
 288                 leader.initiateCaptureSnapshot(addServer.getNewServerId());
 289
 290                 // schedule the install snapshot timeout timer
 291                 Cancellable installSnapshotTimer = raftContext.getActorSystem().scheduler().scheduleOnce(
 292                         new FiniteDuration(((raftContext.getConfigParams().getElectionTimeOutInterval().toMillis()) * 2),
 293                                 TimeUnit.MILLISECONDS), raftContext.getActor(),
 294                                 new FollowerCatchUpTimeout(addServer.getNewServerId()),
 295                                 raftContext.getActorSystem().dispatcher(), raftContext.getActor());
 296
 297                 currentOperationState = new InstallingSnapshot(getAddServerContext(), installSnapshotTimer);
 298             } else {
 299                 LOG.debug("{}: New follower is non-voting - directly persisting new server configuration",
 300                         raftContext.getId());
 301
 302                 persistNewServerConfiguration(raftActor, getAddServerContext());
 303             }
 304         }
 305     }
 306
 307     /**
 308      * The AddServer operation state for when the catch-up snapshot is being installed. It handles successful
 309      * reply or timeout.
 310      */
 311     private class InstallingSnapshot extends AddServerState {
 312         private final Cancellable installSnapshotTimer;
 313
 314         InstallingSnapshot(AddServerContext addServerContext, Cancellable installSnapshotTimer) {
 315             super(addServerContext);
 316             this.installSnapshotTimer = Preconditions.checkNotNull(installSnapshotTimer);
 317         }
 318
 319         @Override
 320         public void onFollowerCatchupTimeout(RaftActor raftActor, FollowerCatchUpTimeout followerTimeout) {
 321             String serverId = followerTimeout.getNewServerId();
 322
 323             LOG.debug("{}: onFollowerCatchupTimeout: {}", raftContext.getId(), serverId);
 324
 325             AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
 326
 327             // cleanup
 328             raftContext.removePeer(serverId);
 329             leader.removeFollower(serverId);
 330
 331             LOG.warn("{}: Timeout occured for new server {} while installing snapshot", raftContext.getId(), serverId);
 332
 333             operationComplete(raftActor, getAddServerContext(), ServerChangeStatus.TIMEOUT);
 334         }
 335
 336         @Override
 337         public void onUnInitializedFollowerSnapshotReply(RaftActor raftActor, UnInitializedFollowerSnapshotReply reply) {
 338             LOG.debug("{}: onUnInitializedFollowerSnapshotReply: {}", raftContext.getId(), reply);
 339
 340             String followerId = reply.getFollowerId();
 341
 342             // Sanity check to guard against receiving an UnInitializedFollowerSnapshotReply from a prior
 343             // add server operation that timed out.
 344             if(getAddServerContext().getOperation().getNewServerId().equals(followerId)) {
 345                 AbstractLeader leader = (AbstractLeader) raftActor.getCurrentBehavior();
 346                 raftContext.getPeerInfo(followerId).setVotingState(VotingState.VOTING);
 347                 leader.updateMinReplicaCount();
 348
 349                 persistNewServerConfiguration(raftActor, getAddServerContext());
 350
 351                 installSnapshotTimer.cancel();
 352             }
 353         }
 354     }
 355
 356     /**
 357      * Stores context information for a server operation.
 358      *
 359      * @param <T> the operation type
 360      */
 361     private static abstract class ServerOperationContext<T> {
 362         private final T operation;
 363         private final ActorRef clientRequestor;
 364         private final String contextId;
 365
 366         ServerOperationContext(T operation, ActorRef clientRequestor){
 367             this.operation = operation;
 368             this.clientRequestor = clientRequestor;
 369             contextId = UUID.randomUUID().toString();
 370         }
 371
 372         String getContextId() {
 373             return contextId;
 374         }
 375
 376         T getOperation() {
 377             return operation;
 378         }
 379
 380         ActorRef getClientRequestor() {
 381             return clientRequestor;
 382         }
 383
 384         abstract Object newReply(ServerChangeStatus status, String leaderId);
 385
 386         abstract InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support);
 387     }
 388
 389     /**
 390      * Stores context information for an AddServer operation.
 391      */
 392     private static class AddServerContext extends ServerOperationContext<AddServer> {
 393         AddServerContext(AddServer addServer, ActorRef clientRequestor) {
 394             super(addServer, clientRequestor);
 395         }
 396
 397         @Override
 398         Object newReply(ServerChangeStatus status, String leaderId) {
 399             return new AddServerReply(status, leaderId);
 400         }
 401
 402         @Override
 403         InitialOperationState newInitialOperationState(RaftActorServerConfigurationSupport support) {
 404             return support.new InitialAddServerState(this);
 405         }
 406     }
 407 }