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(appendEntries.toString());
  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                         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                         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                         , 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(), 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(
 166                     "Number of entries to be appended = {}", appendEntries.getEntries().size()
 167                 );
 168             }
 169
 170             // 3. If an existing entry conflicts with a new one (same index
 171             // but different terms), delete the existing entry and all that
 172             // follow it (§5.3)
 173             int addEntriesFrom = 0;
 174             if (context.getReplicatedLog().size() > 0) {
 175
 176                 // Find the entry up until which the one that is not in the follower's log
 177                 for (int i = 0;i < appendEntries.getEntries().size(); i++, addEntriesFrom++) {
 178                     ReplicatedLogEntry matchEntry = appendEntries.getEntries().get(i);
 179                     ReplicatedLogEntry newEntry = context.getReplicatedLog().get(matchEntry.getIndex());
 180
 181                     if (newEntry == null) {
 182                         //newEntry not found in the log
 183                         break;
 184                     }
 185
 186                     if (newEntry.getTerm() == matchEntry
 187                         .getTerm()) {
 188                         continue;
 189                     }
 190
 191                     if(LOG.isDebugEnabled()) {
 192                         LOG.debug(
 193                             "Removing entries from log starting at {}", matchEntry.getIndex()
 194                         );
 195                     }
 196
 197                     // Entries do not match so remove all subsequent entries
 198                     context.getReplicatedLog()
 199                         .removeFromAndPersist(matchEntry.getIndex());
 200                     break;
 201                 }
 202             }
 203
 204             if(LOG.isDebugEnabled()) {
 205                 LOG.debug("After cleanup entries to be added from = {}", (addEntriesFrom + lastIndex())
 206                 );
 207             }
 208
 209             // 4. Append any new entries not already in the log
 210             for (int i = addEntriesFrom;
 211                  i < appendEntries.getEntries().size(); i++) {
 212
 213                 if(LOG.isDebugEnabled()) {
 214                     LOG.debug("Append entry to log {}", appendEntries.getEntries().get(i).getData());
 215                 }
 216                 context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i));
 217             }
 218
 219             if(LOG.isDebugEnabled()) {
 220                 LOG.debug("Log size is now {}", context.getReplicatedLog().size());
 221             }
 222         }
 223
 224
 225         // 5. If leaderCommit > commitIndex, set commitIndex =
 226         // min(leaderCommit, index of last new entry)
 227
 228         long prevCommitIndex = context.getCommitIndex();
 229
 230         context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(),
 231             context.getReplicatedLog().lastIndex()));
 232
 233         if (prevCommitIndex != context.getCommitIndex()) {
 234             if(LOG.isDebugEnabled()) {
 235                 LOG.debug("Commit index set to {}", context.getCommitIndex());
 236             }
 237         }
 238
 239         // If commitIndex > lastApplied: increment lastApplied, apply
 240         // log[lastApplied] to state machine (§5.3)
 241         // check if there are any entries to be applied. last-applied can be equal to last-index
 242         if (appendEntries.getLeaderCommit() > context.getLastApplied() &&
 243             context.getLastApplied() < lastIndex()) {
 244             if(LOG.isDebugEnabled()) {
 245                 LOG.debug("applyLogToStateMachine, " +
 246                         "appendEntries.getLeaderCommit():{}," +
 247                         "context.getLastApplied():{}, lastIndex():{}",
 248                     appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex()
 249                 );
 250             }
 251
 252             applyLogToStateMachine(appendEntries.getLeaderCommit());
 253         }
 254
 255         sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
 256             lastIndex(), lastTerm()), actor());
 257
 258         return this;
 259     }
 260
 261     @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
 262         AppendEntriesReply appendEntriesReply) {
 263         return this;
 264     }
 265
 266     @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
 267         RequestVoteReply requestVoteReply) {
 268         return this;
 269     }
 270
 271     @Override public RaftState state() {
 272         return RaftState.Follower;
 273     }
 274
 275     @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
 276
 277         Object message = fromSerializableMessage(originalMessage);
 278
 279         if (message instanceof RaftRPC) {
 280             RaftRPC rpc = (RaftRPC) message;
 281             // If RPC request or response contains term T > currentTerm:
 282             // set currentTerm = T, convert to follower (§5.1)
 283             // This applies to all RPC messages and responses
 284             if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
 285                 context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
 286             }
 287         }
 288
 289         if (message instanceof ElectionTimeout) {
 290             return switchBehavior(new Candidate(context));
 291
 292         } else if (message instanceof InstallSnapshot) {
 293             InstallSnapshot installSnapshot = (InstallSnapshot) message;
 294             handleInstallSnapshot(sender, installSnapshot);
 295         }
 296
 297         scheduleElection(electionDuration());
 298
 299         return super.handleMessage(sender, message);
 300     }
 301
 302     private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) {
 303
 304         if(LOG.isDebugEnabled()) {
 305             LOG.debug("InstallSnapshot received by follower " +
 306                     "datasize:{} , Chunk:{}/{}", installSnapshot.getData().size(),
 307                 installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks()
 308             );
 309         }
 310
 311         if(snapshotTracker == null){
 312             snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks());
 313         }
 314
 315         try {
 316             if(snapshotTracker.addChunk(installSnapshot.getChunkIndex(), installSnapshot.getData(),
 317                     installSnapshot.getLastChunkHashCode())){
 318                 Snapshot snapshot = Snapshot.create(snapshotTracker.getSnapshot(),
 319                         new ArrayList<ReplicatedLogEntry>(),
 320                         installSnapshot.getLastIncludedIndex(),
 321                         installSnapshot.getLastIncludedTerm(),
 322                         installSnapshot.getLastIncludedIndex(),
 323                         installSnapshot.getLastIncludedTerm());
 324
 325                 actor().tell(new ApplySnapshot(snapshot), actor());
 326
 327                 snapshotTracker = null;
 328
 329             }
 330
 331             sender.tell(new InstallSnapshotReply(
 332                     currentTerm(), context.getId(), installSnapshot.getChunkIndex(),
 333                     true), actor());
 334
 335         } catch (SnapshotTracker.InvalidChunkException e) {
 336
 337             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 338                     -1, false), actor());
 339             snapshotTracker = null;
 340
 341         } catch (Exception e){
 342
 343             LOG.error(e, "Exception in InstallSnapshot of follower:");
 344             //send reply with success as false. The chunk will be sent again on failure
 345             sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
 346                     installSnapshot.getChunkIndex(), false), actor());
 347
 348         }
 349     }
 350
 351     @Override public void close() throws Exception {
 352         stopElection();
 353     }
 354
 355     @VisibleForTesting
 356     ByteString getSnapshotChunksCollected(){
 357         return snapshotTracker != null ? snapshotTracker.getCollectedChunks() : ByteString.EMPTY;
 358     }
 359
 360
 361 }