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