opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java

   1 package org.opendaylight.controller.cluster.raft.behaviors;
   2
   3 import static org.junit.Assert.assertEquals;
   4 import static org.junit.Assert.assertTrue;
   5 import akka.actor.ActorRef;
   6 import akka.actor.Props;
   7 import akka.testkit.TestActorRef;
   8 import com.google.protobuf.ByteString;
   9 import java.util.ArrayList;
  10 import java.util.Arrays;
  11 import java.util.HashMap;
  12 import java.util.List;
  13 import org.junit.After;
  14 import org.junit.Assert;
  15 import org.junit.Test;
  16 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
  17 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  18 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  19 import org.opendaylight.controller.cluster.raft.Snapshot;
  20 import org.opendaylight.controller.cluster.raft.TestActorFactory;
  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.RequestVote;
  28 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  29 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  30
  31 public class FollowerTest extends AbstractRaftActorBehaviorTest {
  32
  33     private final TestActorFactory actorFactory = new TestActorFactory(getSystem());
  34
  35     private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
  36             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
  37
  38     private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
  39             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
  40
  41     private RaftActorBehavior follower;
  42
  43     @After
  44     public void tearDown() throws Exception {
  45         if(follower != null) {
  46             follower.close();
  47         }
  48
  49         actorFactory.close();
  50     }
  51
  52     @Override
  53     protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
  54         return new Follower(actorContext);
  55     }
  56
  57     @Override
  58     protected  MockRaftActorContext createActorContext() {
  59         return createActorContext(followerActor);
  60     }
  61
  62     @Override
  63     protected  MockRaftActorContext createActorContext(ActorRef actorRef){
  64         return new MockRaftActorContext("follower", getSystem(), actorRef);
  65     }
  66
  67     @Test
  68     public void testThatAnElectionTimeoutIsTriggered(){
  69         MockRaftActorContext actorContext = createActorContext();
  70         follower = new Follower(actorContext);
  71
  72         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class,
  73                 actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
  74     }
  75
  76     @Test
  77     public void testHandleElectionTimeout(){
  78         logStart("testHandleElectionTimeout");
  79
  80         follower = new Follower(createActorContext());
  81
  82         RaftActorBehavior raftBehavior = follower.handleMessage(followerActor, new ElectionTimeout());
  83
  84         assertTrue(raftBehavior instanceof Candidate);
  85     }
  86
  87     @Test
  88     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull(){
  89         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
  90
  91         RaftActorContext context = createActorContext();
  92         long term = 1000;
  93         context.getTermInformation().update(term, null);
  94
  95         follower = createBehavior(context);
  96
  97         follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999));
  98
  99         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
 100
 101         assertEquals("isVoteGranted", true, reply.isVoteGranted());
 102         assertEquals("getTerm", term, reply.getTerm());
 103     }
 104
 105     @Test
 106     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId(){
 107         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
 108
 109         RaftActorContext context = createActorContext();
 110         long term = 1000;
 111         context.getTermInformation().update(term, "test");
 112
 113         follower = createBehavior(context);
 114
 115         follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999));
 116
 117         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
 118
 119         assertEquals("isVoteGranted", false, reply.isVoteGranted());
 120     }
 121
 122     /**
 123      * This test verifies that when an AppendEntries RPC is received by a RaftActor
 124      * with a commitIndex that is greater than what has been applied to the
 125      * state machine of the RaftActor, the RaftActor applies the state and
 126      * sets it current applied state to the commitIndex of the sender.
 127      *
 128      * @throws Exception
 129      */
 130     @Test
 131     public void testHandleAppendEntriesWithNewerCommitIndex() throws Exception {
 132         logStart("testHandleAppendEntriesWithNewerCommitIndex");
 133
 134         MockRaftActorContext context = createActorContext();
 135
 136         context.setLastApplied(100);
 137         setLastLogEntry(context, 1, 100,
 138                 new MockRaftActorContext.MockPayload(""));
 139         context.getReplicatedLog().setSnapshotIndex(99);
 140
 141         List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList(
 142                 newReplicatedLogEntry(2, 101, "foo"));
 143
 144         // The new commitIndex is 101
 145         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
 146
 147         follower = createBehavior(context);
 148         follower.handleMessage(leaderActor, appendEntries);
 149
 150         assertEquals("getLastApplied", 101L, context.getLastApplied());
 151     }
 152
 153     /**
 154      * This test verifies that when an AppendEntries is received a specific prevLogTerm
 155      * which does not match the term that is in RaftActors log entry at prevLogIndex
 156      * then the RaftActor does not change it's state and it returns a failure.
 157      *
 158      * @throws Exception
 159      */
 160     @Test
 161     public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
 162         logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm");
 163
 164         MockRaftActorContext context = createActorContext();
 165
 166         // First set the receivers term to lower number
 167         context.getTermInformation().update(95, "test");
 168
 169         // AppendEntries is now sent with a bigger term
 170         // this will set the receivers term to be the same as the sender's term
 171         AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, null, 101, -1);
 172
 173         follower = createBehavior(context);
 174
 175         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 176
 177         Assert.assertSame(follower, newBehavior);
 178
 179         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
 180                 AppendEntriesReply.class);
 181
 182         assertEquals("isSuccess", false, reply.isSuccess());
 183     }
 184
 185     /**
 186      * This test verifies that when a new AppendEntries message is received with
 187      * new entries and the logs of the sender and receiver match that the new
 188      * entries get added to the log and the log is incremented by the number of
 189      * entries received in appendEntries
 190      *
 191      * @throws Exception
 192      */
 193     @Test
 194     public void testHandleAppendEntriesAddNewEntries() {
 195         logStart("testHandleAppendEntriesAddNewEntries");
 196
 197         MockRaftActorContext context = createActorContext();
 198
 199         // First set the receivers term to lower number
 200         context.getTermInformation().update(1, "test");
 201
 202         // Prepare the receivers log
 203         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 204         log.append(newReplicatedLogEntry(1, 0, "zero"));
 205         log.append(newReplicatedLogEntry(1, 1, "one"));
 206         log.append(newReplicatedLogEntry(1, 2, "two"));
 207
 208         context.setReplicatedLog(log);
 209
 210         // Prepare the entries to be sent with AppendEntries
 211         List<ReplicatedLogEntry> entries = new ArrayList<>();
 212         entries.add(newReplicatedLogEntry(1, 3, "three"));
 213         entries.add(newReplicatedLogEntry(1, 4, "four"));
 214
 215         // Send appendEntries with the same term as was set on the receiver
 216         // before the new behavior was created (1 in this case)
 217         // This will not work for a Candidate because as soon as a Candidate
 218         // is created it increments the term
 219         AppendEntries appendEntries = new AppendEntries(1, "leader-1", 2, 1, entries, 4, -1);
 220
 221         follower = createBehavior(context);
 222
 223         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 224
 225         Assert.assertSame(follower, newBehavior);
 226
 227         assertEquals("Next index", 5, log.last().getIndex() + 1);
 228         assertEquals("Entry 3", entries.get(0), log.get(3));
 229         assertEquals("Entry 4", entries.get(1), log.get(4));
 230
 231         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
 232     }
 233
 234     /**
 235      * This test verifies that when a new AppendEntries message is received with
 236      * new entries and the logs of the sender and receiver are out-of-sync that
 237      * the log is first corrected by removing the out of sync entries from the
 238      * log and then adding in the new entries sent with the AppendEntries message
 239      */
 240     @Test
 241     public void testHandleAppendEntriesCorrectReceiverLogEntries() {
 242         logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
 243
 244         MockRaftActorContext context = createActorContext();
 245
 246         // First set the receivers term to lower number
 247         context.getTermInformation().update(1, "test");
 248
 249         // Prepare the receivers log
 250         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 251         log.append(newReplicatedLogEntry(1, 0, "zero"));
 252         log.append(newReplicatedLogEntry(1, 1, "one"));
 253         log.append(newReplicatedLogEntry(1, 2, "two"));
 254
 255         context.setReplicatedLog(log);
 256
 257         // Prepare the entries to be sent with AppendEntries
 258         List<ReplicatedLogEntry> entries = new ArrayList<>();
 259         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
 260         entries.add(newReplicatedLogEntry(2, 3, "three"));
 261
 262         // Send appendEntries with the same term as was set on the receiver
 263         // before the new behavior was created (1 in this case)
 264         // This will not work for a Candidate because as soon as a Candidate
 265         // is created it increments the term
 266         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1);
 267
 268         follower = createBehavior(context);
 269
 270         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 271
 272         Assert.assertSame(follower, newBehavior);
 273
 274         // The entry at index 2 will be found out-of-sync with the leader
 275         // and will be removed
 276         // Then the two new entries will be added to the log
 277         // Thus making the log to have 4 entries
 278         assertEquals("Next index", 4, log.last().getIndex() + 1);
 279         //assertEquals("Entry 2", entries.get(0), log.get(2));
 280
 281         assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
 282
 283         // Check that the entry at index 2 has the new data
 284         assertEquals("Entry 2", entries.get(0), log.get(2));
 285
 286         assertEquals("Entry 3", entries.get(1), log.get(3));
 287
 288         expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
 289     }
 290
 291     @Test
 292     public void testHandleAppendEntriesPreviousLogEntryMissing(){
 293         logStart("testHandleAppendEntriesPreviousLogEntryMissing");
 294
 295         MockRaftActorContext context = createActorContext();
 296
 297         // Prepare the receivers log
 298         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 299         log.append(newReplicatedLogEntry(1, 0, "zero"));
 300         log.append(newReplicatedLogEntry(1, 1, "one"));
 301         log.append(newReplicatedLogEntry(1, 2, "two"));
 302
 303         context.setReplicatedLog(log);
 304
 305         // Prepare the entries to be sent with AppendEntries
 306         List<ReplicatedLogEntry> entries = new ArrayList<>();
 307         entries.add(newReplicatedLogEntry(1, 4, "four"));
 308
 309         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1);
 310
 311         follower = createBehavior(context);
 312
 313         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 314
 315         Assert.assertSame(follower, newBehavior);
 316
 317         expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
 318     }
 319
 320     @Test
 321     public void testHandleAppendEntriesWithExistingLogEntry() {
 322         logStart("testHandleAppendEntriesWithExistingLogEntry");
 323
 324         MockRaftActorContext context = createActorContext();
 325
 326         context.getTermInformation().update(1, "test");
 327
 328         // Prepare the receivers log
 329         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 330         log.append(newReplicatedLogEntry(1, 0, "zero"));
 331         log.append(newReplicatedLogEntry(1, 1, "one"));
 332
 333         context.setReplicatedLog(log);
 334
 335         // Send the last entry again.
 336         List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
 337
 338         follower = createBehavior(context);
 339
 340         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1));
 341
 342         assertEquals("Next index", 2, log.last().getIndex() + 1);
 343         assertEquals("Entry 1", entries.get(0), log.get(1));
 344
 345         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
 346
 347         // Send the last entry again and also a new one.
 348
 349         entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
 350
 351         leaderActor.underlyingActor().clear();
 352         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1));
 353
 354         assertEquals("Next index", 3, log.last().getIndex() + 1);
 355         assertEquals("Entry 1", entries.get(0), log.get(1));
 356         assertEquals("Entry 2", entries.get(1), log.get(2));
 357
 358         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
 359     }
 360
 361     @Test
 362     public void testHandleAppendAfterInstallingSnapshot(){
 363         logStart("testHandleAppendAfterInstallingSnapshot");
 364
 365         MockRaftActorContext context = createActorContext();
 366
 367         // Prepare the receivers log
 368         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 369
 370         // Set up a log as if it has been snapshotted
 371         log.setSnapshotIndex(3);
 372         log.setSnapshotTerm(1);
 373
 374         context.setReplicatedLog(log);
 375
 376         // Prepare the entries to be sent with AppendEntries
 377         List<ReplicatedLogEntry> entries = new ArrayList<>();
 378         entries.add(newReplicatedLogEntry(1, 4, "four"));
 379
 380         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3);
 381
 382         follower = createBehavior(context);
 383
 384         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 385
 386         Assert.assertSame(follower, newBehavior);
 387
 388         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
 389     }
 390
 391
 392     /**
 393      * This test verifies that when InstallSnapshot is received by
 394      * the follower its applied correctly.
 395      *
 396      * @throws Exception
 397      */
 398     @Test
 399     public void testHandleInstallSnapshot() throws Exception {
 400         logStart("testHandleInstallSnapshot");
 401
 402         MockRaftActorContext context = createActorContext();
 403
 404         follower = createBehavior(context);
 405
 406         HashMap<String, String> followerSnapshot = new HashMap<>();
 407         followerSnapshot.put("1", "A");
 408         followerSnapshot.put("2", "B");
 409         followerSnapshot.put("3", "C");
 410
 411         ByteString bsSnapshot  = toByteString(followerSnapshot);
 412         int offset = 0;
 413         int snapshotLength = bsSnapshot.size();
 414         int chunkSize = 50;
 415         int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
 416         int lastIncludedIndex = 1;
 417         int chunkIndex = 1;
 418         InstallSnapshot lastInstallSnapshot = null;
 419
 420         for(int i = 0; i < totalChunks; i++) {
 421             ByteString chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
 422             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
 423                     chunkData, chunkIndex, totalChunks);
 424             follower.handleMessage(leaderActor, lastInstallSnapshot);
 425             offset = offset + 50;
 426             lastIncludedIndex++;
 427             chunkIndex++;
 428         }
 429
 430         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
 431                 ApplySnapshot.class);
 432         Snapshot snapshot = applySnapshot.getSnapshot();
 433         assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
 434         assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
 435                 snapshot.getLastAppliedTerm());
 436         assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
 437                 snapshot.getLastAppliedIndex());
 438         assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
 439         Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), snapshot.getState());
 440
 441         List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
 442                 leaderActor, InstallSnapshotReply.class);
 443         assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
 444
 445         chunkIndex = 1;
 446         for(InstallSnapshotReply reply: replies) {
 447             assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
 448             assertEquals("getTerm", 1, reply.getTerm());
 449             assertEquals("isSuccess", true, reply.isSuccess());
 450             assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
 451         }
 452
 453         Assert.assertNull("Expected null SnapshotTracker", ((Follower)follower).getSnapshotTracker());
 454     }
 455
 456     @Test
 457     public void testHandleOutOfSequenceInstallSnapshot() throws Exception {
 458         logStart("testHandleOutOfSequenceInstallSnapshot");
 459
 460         MockRaftActorContext context = createActorContext();
 461
 462         follower = createBehavior(context);
 463
 464         HashMap<String, String> followerSnapshot = new HashMap<>();
 465         followerSnapshot.put("1", "A");
 466         followerSnapshot.put("2", "B");
 467         followerSnapshot.put("3", "C");
 468
 469         ByteString bsSnapshot = toByteString(followerSnapshot);
 470
 471         InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
 472                 getNextChunk(bsSnapshot, 10, 50), 3, 3);
 473         follower.handleMessage(leaderActor, installSnapshot);
 474
 475         InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
 476                 InstallSnapshotReply.class);
 477
 478         assertEquals("isSuccess", false, reply.isSuccess());
 479         assertEquals("getChunkIndex", -1, reply.getChunkIndex());
 480         assertEquals("getTerm", 1, reply.getTerm());
 481         assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
 482
 483         Assert.assertNull("Expected null SnapshotTracker", ((Follower)follower).getSnapshotTracker());
 484     }
 485
 486     public ByteString getNextChunk (ByteString bs, int offset, int chunkSize){
 487         int snapshotLength = bs.size();
 488         int start = offset;
 489         int size = chunkSize;
 490         if (chunkSize > snapshotLength) {
 491             size = snapshotLength;
 492         } else {
 493             if ((start + chunkSize) > snapshotLength) {
 494                 size = snapshotLength - start;
 495             }
 496         }
 497         return bs.substring(start, start + size);
 498     }
 499
 500     private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
 501             String expFollowerId, long expLogLastTerm, long expLogLastIndex) {
 502
 503         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
 504                 AppendEntriesReply.class);
 505
 506         assertEquals("isSuccess", expSuccess, reply.isSuccess());
 507         assertEquals("getTerm", expTerm, reply.getTerm());
 508         assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
 509         assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
 510         assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
 511     }
 512
 513     private ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
 514         return new MockRaftActorContext.MockReplicatedLogEntry(term, index,
 515                 new MockRaftActorContext.MockPayload(data));
 516     }
 517 }