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 com.google.protobuf.ByteString;
  14 import java.util.ArrayList;
  15 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  16 import org.opendaylight.controller.cluster.raft.RaftState;
  17 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  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.RequestVoteReply;
  27
  28 /**
  29  * The behavior of a RaftActor in the Follower state
  30  * <p/>
  31  * <ul>
  32  * <li> Respond to RPCs from candidates and leaders
  33  * <li> If election timeout elapses without receiving AppendEntries
  34  * RPC from current leader or granting vote to candidate:
  35  * convert to candidate
  36  * </ul>
  37  */
  38 public class Follower extends AbstractRaftActorBehavior {
  39
  40     private SnapshotTracker snapshotTracker = null;
  41
  42     public Follower(RaftActorContext context) {
  43         super(context);
  44
  45         scheduleElection(electionDuration());
  46     }
  47
  48     private boolean isLogEntryPresent(long index){
  49         if(index == context.getReplicatedLog().getSnapshotIndex()){
  50             return true;
  51         }
  52
  53         ReplicatedLogEntry previousEntry = context.getReplicatedLog()
  54                 .get(index);
  55
  56         return previousEntry != null;
  57
  58     }
  59
  60     private long getLogEntryTerm(long index){
  61         if(index == context.getReplicatedLog().getSnapshotIndex()){
  62             return context.getReplicatedLog().getSnapshotTerm();
  63         }
  64
  65         ReplicatedLogEntry previousEntry = context.getReplicatedLog()
  66                 .get(index);
  67
  68         if(previousEntry != null){
  69             return previousEntry.getTerm();
  70         }
  71
  72         return -1;
  73     }
  74
  75     @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
  76                                                               AppendEntries appendEntries) {
  77
  78         if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
  79             if(LOG.isDebugEnabled()) {
  80                 LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
  81             }
  82         }
  83
  84         // TODO : Refactor this method into a bunch of smaller methods
  85         // to make it easier to read. Before refactoring ensure tests
  86         // cover the code properly
  87
  88         // 1. Reply false if term < currentTerm (§5.1)
  89         // This is handled in the appendEntries method of the base class
  90
  91         // If we got here then we do appear to be talking to the leader
  92         leaderId = appendEntries.getLeaderId();
  93
  94         // 2. Reply false if log doesn’t contain an entry at prevLogIndex
  95         // whose term matches prevLogTerm (§5.3)
  96
  97         long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex());
  98         boolean prevEntryPresent = isLogEntryPresent(appendEntries.getPrevLogIndex());
  99
 100
 101         boolean outOfSync = true;
 102
 103         // First check if the logs are in sync or not
 104         if (lastIndex() == -1
 105                 && appendEntries.getPrevLogIndex() != -1) {
 106
 107             // The follower's log is out of sync because the leader does have
 108             // an entry at prevLogIndex and this follower has no entries in
 109             // it's log.
 110
 111             if(LOG.isDebugEnabled()) {
 112                 LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
 113                         context.getId(), appendEntries.getPrevLogIndex());
 114             }
 115
 116         } else if (lastIndex() > -1
 117                 && appendEntries.getPrevLogIndex() != -1
 118                 && !prevEntryPresent) {
 119
 120             // The follower's log is out of sync because the Leader's
 121             // prevLogIndex entry was not found in it's log
 122
 123             if(LOG.isDebugEnabled()) {
 124                 LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
 125                         context.getId(), appendEntries.getPrevLogIndex());
 126             }
 127
 128         } else if (lastIndex() > -1
 129                 && prevEntryPresent
 130                 && prevLogTerm != appendEntries.getPrevLogTerm()) {
 131
 132             // The follower's log is out of sync because the Leader's
 133             // prevLogIndex entry does exist in the follower's log but it has
 134             // a different term in it
 135
 136             if (LOG.isDebugEnabled()) {
 137                 LOG.debug(
 138                         "{}: Cannot append entries because previous entry term {}  is not equal to append entries prevLogTerm {}"
 139                         , context.getId(), prevLogTerm
 140                         , appendEntries.getPrevLogTerm());
 141             }
 142         } else {
 143             outOfSync = false;
 144         }
 145
 146         if (outOfSync) {
 147             // We found that the log was out of sync so just send a negative
 148             // reply and return
 149             if(LOG.isDebugEnabled()) {
 150                 LOG.debug("{}: Follower ({}) is out-of-sync, " +
 151                         "so sending negative reply, lastIndex():{}, lastTerm():{}",
 152                         context.getId(), context.getId(), lastIndex(), lastTerm()
 153                 );
 154             }
 155             sender.tell(
 156                 new AppendEntriesReply(context.getId(), currentTerm(), false,
 157                     lastIndex(), lastTerm()), actor()
 158             );
 159             return this;
 160         }
 161
 162         if (appendEntries.getEntries() != null
 163             && appendEntries.getEntries().size() > 0) {
 164             if(LOG.isDebugEnabled()) {
 165                 LOG.debug("{}: Number of entries to be appended = {}", context.getId(),
 166                         appendEntries.getEntries().size());
 167             }
 168
 169             // 3. If an existing entry conflicts with a new one (same index
 170             // but different terms), delete the existing entry and all that
 171             // follow it (§5.3)
 172             int addEntriesFrom = 0;
 173             if (context.getReplicatedLog().size() > 0) {
 174
 175                 // Find the entry up until which the one that is not in the follower's log
 176                 for (int i = 0;i < appendEntries.getEntries().size(); i++, addEntriesFrom++) {
 177                     ReplicatedLogEntry matchEntry = appendEntries.getEntries().get(i);
 178                     ReplicatedLogEntry newEntry = context.getReplicatedLog().get(matchEntry.getIndex());
 179
 180                     if (newEntry == null) {
 181                         //newEntry not found in the log
 182                         break;
 183                     }
 184
 185                     if (newEntry.getTerm() == matchEntry
 186                         .getTerm()) {
 187                         continue;
 188                     }
 189
 190                     if(LOG.isDebugEnabled()) {
 191                         LOG.debug("{}: Removing entries from log starting at {}", context.getId(),
 192                                 matchEntry.getIndex());
 193                     }
 194
 195                     // Entries do not match so remove all subsequent entries
 196                     context.getReplicatedLog()
 197                         .removeFromAndPersist(matchEntry.getIndex());
 198                     break;
 199                 }
 200             }
 201
 202             if(LOG.isDebugEnabled()) {
 203                 LOG.debug("{}: After cleanup entries to be added from = {}", context.getId(),
 204                         (addEntriesFrom + lastIndex()));
 205             }
 206
 207             // 4. Append any new entries not already in the log
 208             for (int i = addEntriesFrom;
 209                  i < appendEntries.getEntries().size(); i++) {
 210
 211                 if(LOG.isDebugEnabled()) {
 212                     LOG.debug("{}: Append entry to log {}", context.getId(),
 213                             appendEntries.getEntries().get(i).getData());
 214                 }
 215                 context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i));
 216             }
 217
 218             if(LOG.isDebugEnabled()) {
 219                 LOG.debug("{}: Log size is now {}", context.getId(), context.getReplicatedLog().size());
 220             }
 221         }
 222
 223
 224         // 5. If leaderCommit > commitIndex, set commitIndex =
 225         // min(leaderCommit, index of last new entry)
 226
 227         long prevCommitIndex = context.getCommitIndex();
 228
 229         context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(),
 230             context.getReplicatedLog().lastIndex()));
 231
 232         if (prevCommitIndex != context.getCommitIndex()) {
 233             if(LOG.isDebugEnabled()) {
 234                 LOG.debug("{}: Commit index set to {}", context.getId(), context.getCommitIndex());
 235             }
 236         }
 237
 238         // If commitIndex > lastApplied: increment lastApplied, apply
 239         // log[lastApplied] to state machine (§5.3)
 240         // check if there are any entries to be applied. last-applied can be equal to last-index
 241         if (appendEntries.getLeaderCommit() > context.getLastApplied() &&
 242             context.getLastApplied() < lastIndex()) {
 243             if(LOG.isDebugEnabled()) {
 244                 LOG.debug("{}: applyLogToStateMachine, " +
 245                         "appendEntries.getLeaderCommit():{}," +
 246                         "context.getLastApplied():{}, lastIndex():{}", context.getId(),
 247                     appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex()
 248                 );
 249             }
 250
 251             applyLogToStateMachine(appendEntries.getLeaderCommit());
 252         }
 253
 254         sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
 255             lastIndex(), lastTerm()), actor());
 256
 257         return this;
 258     }
 259
 260     @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
 261         AppendEntriesReply appendEntriesReply) {
 262         return this;
 263     }
 264
 265     @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
 266         RequestVoteReply requestVoteReply) {
 267         return this;
 268     }
 269
 270     @Override public RaftState state() {
 271         return RaftState.Follower;
 272     }
 273
 274     @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 275
 276         Object message = fromSerializableMessage(originalMessage);
 277
 278         if (message instanceof RaftRPC) {
 279             RaftRPC rpc = (RaftRPC) message;
 280             // If RPC request or response contains term T > currentTerm:
 281             // set currentTerm = T, convert to follower (§5.1)
 282             // This applies to all RPC messages and responses
 283             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 284                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 285             }
 286         }
 287
 288         if (message instanceof ElectionTimeout) {
 289             return switchBehavior(new Candidate(context));
 290
 291         } else if (message instanceof InstallSnapshot) {
 292             InstallSnapshot installSnapshot = (InstallSnapshot) message;
 293             handleInstallSnapshot(sender, installSnapshot);
 294         }
 295
 296         scheduleElection(electionDuration());
 297
 298         return super.handleMessage(sender, message);
 299     }
 300
 301     private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) {
 302
 303         if(LOG.isDebugEnabled()) {
 304             LOG.debug("{}: InstallSnapshot received by follower " +
 305                     "datasize:{} , Chunk:{}/{}", context.getId(), installSnapshot.getData().size(),
 306                 installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks()
 307             );
 308         }
 309
 310         if(snapshotTracker == null){
 311             snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks());
 312         }
 313
 314         try {
 315             if(snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(),
 316                     installSnapshot.getLastChunkHashCode())){
 317                 Snapshot snapshot = Snapshot.create(snapshotTracker.getSnapshot(),
 318                         new ArrayList<ReplicatedLogEntry>(),
 319                         installSnapshot.getLastIncludedIndex(),
 320                         installSnapshot.getLastIncludedTerm(),
 321                         installSnapshot.getLastIncludedIndex(),
 322                         installSnapshot.getLastIncludedTerm());
 323
 324                 actor().tell(new ApplySnapshot(snapshot), actor());
 325
 326                 snapshotTracker = null;
 327
 328             }
 329
 330             sender.tell(new InstallSnapshotReply(
 331                     currentTerm(), context.getId(), installSnapshot.getChunkIndex(),
 332                     true), actor());
 333
 334         } catch (SnapshotTracker.InvalidChunkException e) {
 335
 336             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 337                     -1, false), actor());
 338             snapshotTracker = null;
 339
 340         } catch (Exception e){
 341             LOG.error(e, "{}: Exception in InstallSnapshot of follower", context.getId());
 342             //send reply with success as false. The chunk will be sent again on failure
 343             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 344                     installSnapshot.getChunkIndex(), false), actor());
 345
 346         }
 347     }
 348
 349     @Override public void close() throws Exception {
 350         stopElection();
 351     }
 352
 353     @VisibleForTesting
 354     ByteString getSnapshotChunksCollected(){
 355         return snapshotTracker != null ? snapshotTracker.getCollectedChunks() : ByteString.EMPTY;
 356     }
 357
 358
 359 }