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

   1 /*
   2  * Copyright (c) 2014 Cisco 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
   9 package org.opendaylight.controller.cluster.raft.behaviors;
  10
  11 import akka.actor.ActorRef;
  12 import com.google.common.annotations.VisibleForTesting;
  13 import java.util.ArrayList;
  14 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  15 import org.opendaylight.controller.cluster.raft.RaftState;
  16 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  17 import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload;
  18 import org.opendaylight.controller.cluster.raft.Snapshot;
  19 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
  20 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
  21 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  22 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  23 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  24 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  25 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  26 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  27 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  28
  29 /**
  30  * The behavior of a RaftActor in the Follower state
  31  * <p/>
  32  * <ul>
  33  * <li> Respond to RPCs from candidates and leaders
  34  * <li> If election timeout elapses without receiving AppendEntries
  35  * RPC from current leader or granting vote to candidate:
  36  * convert to candidate
  37  * </ul>
  38  */
  39 public class Follower extends AbstractRaftActorBehavior {
  40
  41     private SnapshotTracker snapshotTracker = null;
  42
  43     private final SyncStatusTracker initialSyncStatusTracker;
  44
  45     private static final int SYNC_THRESHOLD = 10;
  46
  47     public Follower(RaftActorContext context) {
  48         this(context, null);
  49     }
  50
  51     public Follower(RaftActorContext context, String initialLeaderId) {
  52         super(context, RaftState.Follower);
  53         leaderId = initialLeaderId;
  54
  55         initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), SYNC_THRESHOLD);
  56
  57         if(canStartElection()) {
  58             if (context.getPeerIds().isEmpty() && getLeaderId() == null) {
  59                 actor().tell(ELECTION_TIMEOUT, actor());
  60             } else {
  61                 scheduleElection(electionDuration());
  62             }
  63         }
  64
  65     }
  66
  67     private boolean isLogEntryPresent(long index){
  68         if(index == context.getReplicatedLog().getSnapshotIndex()){
  69             return true;
  70         }
  71
  72         ReplicatedLogEntry previousEntry = context.getReplicatedLog()
  73                 .get(index);
  74
  75         return previousEntry != null;
  76
  77     }
  78
  79     private long getLogEntryTerm(long index){
  80         if(index == context.getReplicatedLog().getSnapshotIndex()){
  81             return context.getReplicatedLog().getSnapshotTerm();
  82         }
  83
  84         ReplicatedLogEntry previousEntry = context.getReplicatedLog()
  85                 .get(index);
  86
  87         if(previousEntry != null){
  88             return previousEntry.getTerm();
  89         }
  90
  91         return -1;
  92     }
  93
  94     private void updateInitialSyncStatus(long currentLeaderCommit, String leaderId){
  95         initialSyncStatusTracker.update(leaderId, currentLeaderCommit, context.getCommitIndex());
  96     }
  97
  98     @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
  99                                                               AppendEntries appendEntries) {
 100
 101         int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
 102         if(LOG.isTraceEnabled()) {
 103             LOG.trace("{}: handleAppendEntries: {}", logName(), appendEntries);
 104         } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
 105             LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
 106         }
 107
 108         // TODO : Refactor this method into a bunch of smaller methods
 109         // to make it easier to read. Before refactoring ensure tests
 110         // cover the code properly
 111
 112         if (snapshotTracker != null || context.getSnapshotManager().isApplying()) {
 113             // if snapshot install is in progress, follower should just acknowledge append entries with a reply.
 114             AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), true,
 115                     lastIndex(), lastTerm(), context.getPayloadVersion());
 116
 117             if(LOG.isDebugEnabled()) {
 118                 LOG.debug("{}: snapshot install is in progress, replying immediately with {}", logName(), reply);
 119             }
 120             sender.tell(reply, actor());
 121
 122             return this;
 123         }
 124
 125         // If we got here then we do appear to be talking to the leader
 126         leaderId = appendEntries.getLeaderId();
 127
 128         setLeaderPayloadVersion(appendEntries.getPayloadVersion());
 129
 130         updateInitialSyncStatus(appendEntries.getLeaderCommit(), appendEntries.getLeaderId());
 131         // First check if the logs are in sync or not
 132         long lastIndex = lastIndex();
 133
 134         if (isOutOfSync(appendEntries)) {
 135             // We found that the log was out of sync so just send a negative
 136             // reply and return
 137
 138             LOG.debug("{}: Follower is out-of-sync, so sending negative reply, lastIndex: {}, lastTerm: {}",
 139                         logName(), lastIndex, lastTerm());
 140
 141             sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex,
 142                     lastTerm(), context.getPayloadVersion()), actor());
 143             return this;
 144         }
 145
 146         if (appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
 147
 148             LOG.debug("{}: Number of entries to be appended = {}", logName(),
 149                         appendEntries.getEntries().size());
 150
 151             // 3. If an existing entry conflicts with a new one (same index
 152             // but different terms), delete the existing entry and all that
 153             // follow it (§5.3)
 154             int addEntriesFrom = 0;
 155             if (context.getReplicatedLog().size() > 0) {
 156
 157                 // Find the entry up until the one that is not in the follower's log
 158                 for (int i = 0;i < appendEntries.getEntries().size(); i++, addEntriesFrom++) {
 159                     ReplicatedLogEntry matchEntry = appendEntries.getEntries().get(i);
 160                     ReplicatedLogEntry newEntry = context.getReplicatedLog().get(matchEntry.getIndex());
 161
 162                     if (newEntry == null) {
 163                         //newEntry not found in the log
 164                         break;
 165                     }
 166
 167                     if (newEntry.getTerm() == matchEntry.getTerm()) {
 168                         continue;
 169                     }
 170
 171                     if(!context.getRaftPolicy().applyModificationToStateBeforeConsensus()) {
 172
 173                         LOG.debug("{}: Removing entries from log starting at {}", logName(),
 174                                 matchEntry.getIndex());
 175
 176                         // Entries do not match so remove all subsequent entries
 177                         context.getReplicatedLog().removeFromAndPersist(matchEntry.getIndex());
 178                         break;
 179                     } else {
 180                         sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex,
 181                                 lastTerm(), context.getPayloadVersion(), true), actor());
 182                         return this;
 183                     }
 184                 }
 185             }
 186
 187             lastIndex = lastIndex();
 188             LOG.debug("{}: After cleanup entries to be added from = {}", logName(),
 189                         (addEntriesFrom + lastIndex));
 190
 191             // 4. Append any new entries not already in the log
 192             for (int i = addEntriesFrom; i < appendEntries.getEntries().size(); i++) {
 193                 ReplicatedLogEntry entry = appendEntries.getEntries().get(i);
 194
 195                 LOG.debug("{}: Append entry to log {}", logName(), entry.getData());
 196
 197                 context.getReplicatedLog().appendAndPersist(entry);
 198
 199                 if(entry.getData() instanceof ServerConfigurationPayload) {
 200                     context.updatePeerIds((ServerConfigurationPayload)entry.getData());
 201                 }
 202             }
 203
 204             LOG.debug("{}: Log size is now {}", logName(), context.getReplicatedLog().size());
 205         }
 206
 207         // 5. If leaderCommit > commitIndex, set commitIndex =
 208         // min(leaderCommit, index of last new entry)
 209
 210         lastIndex = lastIndex();
 211         long prevCommitIndex = context.getCommitIndex();
 212
 213         context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(), lastIndex));
 214
 215         if (prevCommitIndex != context.getCommitIndex()) {
 216             LOG.debug("{}: Commit index set to {}", logName(), context.getCommitIndex());
 217         }
 218
 219         // If commitIndex > lastApplied: increment lastApplied, apply
 220         // log[lastApplied] to state machine (§5.3)
 221         // check if there are any entries to be applied. last-applied can be equal to last-index
 222         if (appendEntries.getLeaderCommit() > context.getLastApplied() &&
 223             context.getLastApplied() < lastIndex) {
 224             if(LOG.isDebugEnabled()) {
 225                 LOG.debug("{}: applyLogToStateMachine, " +
 226                         "appendEntries.getLeaderCommit(): {}," +
 227                         "context.getLastApplied(): {}, lastIndex(): {}", logName(),
 228                     appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex);
 229             }
 230
 231             applyLogToStateMachine(appendEntries.getLeaderCommit());
 232         }
 233
 234         AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), true,
 235             lastIndex, lastTerm(), context.getPayloadVersion());
 236
 237         if(LOG.isTraceEnabled()) {
 238             LOG.trace("{}: handleAppendEntries returning : {}", logName(), reply);
 239         } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
 240             LOG.debug("{}: handleAppendEntries returning : {}", logName(), reply);
 241         }
 242
 243         sender.tell(reply, actor());
 244
 245         if (!context.getSnapshotManager().isCapturing()) {
 246             super.performSnapshotWithoutCapture(appendEntries.getReplicatedToAllIndex());
 247         }
 248
 249         return this;
 250     }
 251
 252     private boolean isOutOfSync(AppendEntries appendEntries) {
 253
 254         long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex());
 255         boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex());
 256         long lastIndex = lastIndex();
 257         int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
 258         boolean outOfSync = true;
 259
 260         if (lastIndex == -1 && appendEntries.getPrevLogIndex() != -1) {
 261
 262             // The follower's log is out of sync because the leader does have
 263             // an entry at prevLogIndex and this follower has no entries in
 264             // it's log.
 265
 266             LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
 267                         logName(), appendEntries.getPrevLogIndex());
 268         } else if (lastIndex > -1 && appendEntries.getPrevLogIndex() != -1 && !prevEntryPresent) {
 269
 270             // The follower's log is out of sync because the Leader's
 271             // prevLogIndex entry was not found in it's log
 272
 273             LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
 274                         logName(), appendEntries.getPrevLogIndex());
 275         } else if (lastIndex > -1 && prevEntryPresent && prevLogTerm != appendEntries.getPrevLogTerm()) {
 276
 277             // The follower's log is out of sync because the Leader's
 278             // prevLogIndex entry does exist in the follower's log but it has
 279             // a different term in it
 280
 281             LOG.debug(
 282                     "{}: Cannot append entries because previous entry term {}  is not equal to append entries prevLogTerm {}",
 283                     logName(), prevLogTerm, appendEntries.getPrevLogTerm());
 284         } else if(appendEntries.getPrevLogIndex() == -1 && appendEntries.getPrevLogTerm() == -1
 285                 && appendEntries.getReplicatedToAllIndex() != -1
 286                 && !isLogEntryPresent(appendEntries.getReplicatedToAllIndex())
 287                 && !context.getReplicatedLog().isInSnapshot(appendEntries.getReplicatedToAllIndex())) {
 288             // This append entry comes from a leader who has it's log aggressively trimmed and so does not have
 289             // the previous entry in it's in-memory journal
 290
 291             LOG.debug(
 292                     "{}: Cannot append entries because the replicatedToAllIndex {} does not appear to be in the in-memory journal",
 293                     logName(), appendEntries.getReplicatedToAllIndex());
 294         } else if(appendEntries.getPrevLogIndex() == -1 && appendEntries.getPrevLogTerm() == -1
 295                 && appendEntries.getReplicatedToAllIndex() != -1 && numLogEntries > 0
 296                 && !isLogEntryPresent(appendEntries.getEntries().get(0).getIndex() - 1)
 297                 && !context.getReplicatedLog().isInSnapshot(appendEntries.getEntries().get(0).getIndex() - 1)) {
 298             LOG.debug(
 299                     "{}: Cannot append entries because the calculated previousIndex {} was not found in the in-memory journal",
 300                     logName(), appendEntries.getEntries().get(0).getIndex() - 1);
 301         } else {
 302             outOfSync = false;
 303         }
 304         return outOfSync;
 305     }
 306
 307     @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
 308         AppendEntriesReply appendEntriesReply) {
 309         return this;
 310     }
 311
 312     @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
 313         RequestVoteReply requestVoteReply) {
 314         return this;
 315     }
 316
 317     @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 318
 319         Object message = fromSerializableMessage(originalMessage);
 320
 321         if (message instanceof RaftRPC) {
 322             RaftRPC rpc = (RaftRPC) message;
 323             // If RPC request or response contains term T > currentTerm:
 324             // set currentTerm = T, convert to follower (§5.1)
 325             // This applies to all RPC messages and responses
 326             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 327                 LOG.debug("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
 328                         logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
 329
 330                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 331             }
 332         }
 333
 334         if (message instanceof ElectionTimeout) {
 335             if(canStartElection()) {
 336                 LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName());
 337                 return internalSwitchBehavior(RaftState.Candidate);
 338             } else {
 339                 return this;
 340             }
 341
 342         } else if (message instanceof InstallSnapshot) {
 343             InstallSnapshot installSnapshot = (InstallSnapshot) message;
 344             handleInstallSnapshot(sender, installSnapshot);
 345         }
 346
 347         if(message instanceof RaftRPC && (!(message instanceof RequestVote) || (canGrantVote((RequestVote) message)))){
 348             scheduleElection(electionDuration());
 349         }
 350
 351         return super.handleMessage(sender, message);
 352     }
 353
 354     private void handleInstallSnapshot(final ActorRef sender, InstallSnapshot installSnapshot) {
 355
 356         LOG.debug("{}: InstallSnapshot received from leader {}, datasize: {} , Chunk: {}/{}",
 357                     logName(), installSnapshot.getLeaderId(), installSnapshot.getData().size(),
 358                     installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks());
 359
 360         leaderId = installSnapshot.getLeaderId();
 361
 362         if(snapshotTracker == null){
 363             snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks());
 364         }
 365
 366         updateInitialSyncStatus(installSnapshot.getLastIncludedIndex(), installSnapshot.getLeaderId());
 367
 368         try {
 369             final InstallSnapshotReply reply = new InstallSnapshotReply(
 370                     currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true);
 371
 372             if(snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(),
 373                     installSnapshot.getLastChunkHashCode())){
 374                 Snapshot snapshot = Snapshot.create(snapshotTracker.getSnapshot(),
 375                         new ArrayList<ReplicatedLogEntry>(),
 376                         installSnapshot.getLastIncludedIndex(),
 377                         installSnapshot.getLastIncludedTerm(),
 378                         installSnapshot.getLastIncludedIndex(),
 379                         installSnapshot.getLastIncludedTerm(),
 380                         context.getTermInformation().getCurrentTerm(),
 381                         context.getTermInformation().getVotedFor(),
 382                         context.getPeerServerInfo(true));
 383
 384                 ApplySnapshot.Callback applySnapshotCallback = new ApplySnapshot.Callback() {
 385                     @Override
 386                     public void onSuccess() {
 387                         LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply);
 388
 389                         sender.tell(reply, actor());
 390                     }
 391
 392                     @Override
 393                     public void onFailure() {
 394                         sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(), -1, false), actor());
 395                     }
 396                 };
 397
 398                 actor().tell(new ApplySnapshot(snapshot, applySnapshotCallback), actor());
 399
 400                 snapshotTracker = null;
 401             } else {
 402                 LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply);
 403
 404                 sender.tell(reply, actor());
 405             }
 406         } catch (SnapshotTracker.InvalidChunkException e) {
 407             LOG.debug("{}: Exception in InstallSnapshot of follower", logName(), e);
 408
 409             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 410                     -1, false), actor());
 411             snapshotTracker = null;
 412
 413         } catch (Exception e){
 414             LOG.error("{}: Exception in InstallSnapshot of follower", logName(), e);
 415
 416             //send reply with success as false. The chunk will be sent again on failure
 417             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 418                     installSnapshot.getChunkIndex(), false), actor());
 419
 420         }
 421     }
 422
 423     @Override
 424     public void close() throws Exception {
 425         stopElection();
 426     }
 427
 428     @VisibleForTesting
 429     SnapshotTracker getSnapshotTracker(){
 430         return snapshotTracker;
 431     }
 432 }