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

   1 /*
   2  * Copyright (c) 2014, 2015 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 static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertFalse;
  13 import static org.junit.Assert.assertNull;
  14
  15 import akka.actor.ActorRef;
  16 import akka.actor.Props;
  17 import akka.testkit.TestActorRef;
  18 import com.google.protobuf.ByteString;
  19 import java.io.ByteArrayOutputStream;
  20 import java.io.IOException;
  21 import java.io.ObjectOutputStream;
  22 import java.util.ArrayList;
  23 import java.util.Collections;
  24 import java.util.List;
  25 import java.util.Map;
  26 import org.junit.After;
  27 import org.junit.Assert;
  28 import org.junit.Test;
  29 import org.opendaylight.controller.cluster.raft.AbstractActorTest;
  30 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
  31 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  32 import org.opendaylight.controller.cluster.raft.RaftState;
  33 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  34 import org.opendaylight.controller.cluster.raft.TestActorFactory;
  35 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  36 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  37 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  38 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
  39 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  40 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  41 import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
  42 import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
  43 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  44 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  45 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  46 import org.slf4j.LoggerFactory;
  47
  48 public abstract class AbstractRaftActorBehaviorTest<T extends RaftActorBehavior> extends AbstractActorTest {
  49
  50     protected final TestActorFactory actorFactory = new TestActorFactory(getSystem());
  51
  52     private final TestActorRef<MessageCollectorActor> behaviorActor = actorFactory.createTestActor(
  53             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("behavior"));
  54
  55     RaftActorBehavior behavior;
  56
  57     @After
  58     public void tearDown() {
  59         if (behavior != null) {
  60             behavior.close();
  61         }
  62
  63         actorFactory.close();
  64
  65         InMemoryJournal.clear();
  66         InMemorySnapshotStore.clear();
  67     }
  68
  69     /**
  70      * This test checks that when a new Raft RPC message is received with a newer
  71      * term the RaftActor gets into the Follower state.
  72      */
  73     @Test
  74     public void testHandleRaftRPCWithNewerTerm() {
  75         MockRaftActorContext actorContext = createActorContext();
  76
  77         assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor,
  78                 createAppendEntriesWithNewerTerm());
  79
  80         assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor,
  81                 createAppendEntriesReplyWithNewerTerm());
  82
  83         assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor,
  84                 createRequestVoteWithNewerTerm());
  85
  86         assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor,
  87                 createRequestVoteReplyWithNewerTerm());
  88     }
  89
  90
  91     /**
  92      * This test verifies that when an AppendEntries is received with a term that
  93      * is less that the currentTerm of the RaftActor then the RaftActor does not
  94      * change it's state and it responds back with a failure.
  95      */
  96     @Test
  97     public void testHandleAppendEntriesSenderTermLessThanReceiverTerm() {
  98         MockRaftActorContext context = createActorContext();
  99         short payloadVersion = 5;
 100         context.setPayloadVersion(payloadVersion);
 101
 102         // First set the receivers term to a high number (1000)
 103         context.getTermInformation().update(1000, "test");
 104
 105         AppendEntries appendEntries = new AppendEntries(100, "leader-1", 0, 0, Collections.emptyList(), 101, -1,
 106                 (short)4);
 107
 108         behavior = createBehavior(context);
 109
 110         RaftState expected = behavior.state();
 111
 112         RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries);
 113
 114         assertEquals("Raft state", expected, raftBehavior.state());
 115
 116         // Also expect an AppendEntriesReply to be sent where success is false
 117
 118         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(
 119                 behaviorActor, AppendEntriesReply.class);
 120
 121         assertEquals("isSuccess", false, reply.isSuccess());
 122         assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
 123     }
 124
 125
 126     @Test
 127     public void testHandleAppendEntriesAddSameEntryToLog() {
 128         MockRaftActorContext context = createActorContext();
 129
 130         context.getTermInformation().update(2, "test");
 131
 132         // Prepare the receivers log
 133         MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("zero");
 134         setLastLogEntry(context, 2, 0, payload);
 135
 136         List<ReplicatedLogEntry> entries = new ArrayList<>();
 137         entries.add(new SimpleReplicatedLogEntry(0, 2, payload));
 138
 139         final AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 2, -1, (short)0);
 140
 141         behavior = createBehavior(context);
 142
 143         assertFalse("This test should be overridden when testing Candidate", behavior instanceof Candidate);
 144
 145         RaftState expected = behavior.state();
 146
 147         // Check that the behavior does not handle unknwon message
 148         assertNull(behavior.handleMessage(behaviorActor, "unknown"));
 149
 150         RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries);
 151
 152         assertEquals("Raft state", expected, raftBehavior.state());
 153
 154         assertEquals("ReplicatedLog size", 1, context.getReplicatedLog().size());
 155
 156         handleAppendEntriesAddSameEntryToLogReply(behaviorActor);
 157     }
 158
 159     protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor) {
 160         AppendEntriesReply reply = MessageCollectorActor.getFirstMatching(replyActor, AppendEntriesReply.class);
 161         Assert.assertNull("Expected no AppendEntriesReply", reply);
 162     }
 163
 164     /**
 165      * This test verifies that when a RequestVote is received by the RaftActor
 166      * with the senders' log is more up to date than the receiver that the receiver grants
 167      * the vote to the sender.
 168      */
 169     @Test
 170     public void testHandleRequestVoteWhenSenderLogMoreUpToDate() {
 171         MockRaftActorContext context = createActorContext();
 172
 173         behavior = createBehavior(context);
 174
 175         context.getTermInformation().update(1, "test");
 176
 177         behavior.handleMessage(behaviorActor, new RequestVote(context.getTermInformation().getCurrentTerm(),
 178                 "test", 10000, 999));
 179
 180         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor,
 181                 RequestVoteReply.class);
 182         assertEquals("isVoteGranted", true, reply.isVoteGranted());
 183     }
 184
 185     /**
 186      * This test verifies that when a RaftActor receives a RequestVote message
 187      * with a term that is greater than it's currentTerm but a less up-to-date
 188      * log then the receiving RaftActor will not grant the vote to the sender.
 189      */
 190     @Test
 191     public void testHandleRequestVoteWhenSenderLogLessUptoDate() {
 192         MockRaftActorContext context = createActorContext();
 193
 194         behavior = createBehavior(context);
 195
 196         context.getTermInformation().update(1, "test");
 197
 198         int index = 2000;
 199         setLastLogEntry(context, context.getTermInformation().getCurrentTerm(), index,
 200                 new MockRaftActorContext.MockPayload(""));
 201
 202         behavior.handleMessage(behaviorActor, new RequestVote(
 203                 context.getTermInformation().getCurrentTerm(), "test",
 204                 index - 1, context.getTermInformation().getCurrentTerm()));
 205
 206         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor,
 207                 RequestVoteReply.class);
 208         assertEquals("isVoteGranted", false, reply.isVoteGranted());
 209     }
 210
 211
 212
 213     /**
 214      * This test verifies that the receiving RaftActor will not grant a vote
 215      * to a sender if the sender's term is lesser than the currentTerm of the
 216      * recipient RaftActor.
 217      */
 218     @Test
 219     public void testHandleRequestVoteWhenSenderTermLessThanCurrentTerm() {
 220         MockRaftActorContext context = createActorContext();
 221
 222         context.getTermInformation().update(1000, null);
 223
 224         behavior = createBehavior(context);
 225
 226         behavior.handleMessage(behaviorActor, new RequestVote(999, "test", 10000, 999));
 227
 228         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor,
 229                 RequestVoteReply.class);
 230         assertEquals("isVoteGranted", false, reply.isVoteGranted());
 231     }
 232
 233     @Test
 234     public void testPerformSnapshot() {
 235         MockRaftActorContext context = new MockRaftActorContext("test", getSystem(), behaviorActor);
 236         AbstractRaftActorBehavior abstractBehavior =  (AbstractRaftActorBehavior) createBehavior(context);
 237         if (abstractBehavior instanceof Candidate) {
 238             return;
 239         }
 240
 241         context.getTermInformation().update(1, "test");
 242
 243         //log has 1 entry with replicatedToAllIndex = 0, does not do anything, returns the
 244         context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
 245         context.setLastApplied(0);
 246         abstractBehavior.performSnapshotWithoutCapture(0);
 247         assertEquals(-1, abstractBehavior.getReplicatedToAllIndex());
 248         assertEquals(1, context.getReplicatedLog().size());
 249
 250         //2 entries, lastApplied still 0, no purging.
 251         context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
 252         context.setLastApplied(0);
 253         abstractBehavior.performSnapshotWithoutCapture(0);
 254         assertEquals(-1, abstractBehavior.getReplicatedToAllIndex());
 255         assertEquals(2, context.getReplicatedLog().size());
 256
 257         //2 entries, lastApplied still 0, no purging.
 258         context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
 259         context.setLastApplied(1);
 260         abstractBehavior.performSnapshotWithoutCapture(0);
 261         assertEquals(0, abstractBehavior.getReplicatedToAllIndex());
 262         assertEquals(1, context.getReplicatedLog().size());
 263
 264         // 5 entries, lastApplied =2 and replicatedIndex = 3, but since we want to keep the lastapplied, indices 0 and
 265         // 1 will only get purged
 266         context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 5, 1).build());
 267         context.setLastApplied(2);
 268         abstractBehavior.performSnapshotWithoutCapture(3);
 269         assertEquals(1, abstractBehavior.getReplicatedToAllIndex());
 270         assertEquals(3, context.getReplicatedLog().size());
 271
 272         // scenario where Last applied > Replicated to all index (becoz of a slow follower)
 273         context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
 274         context.setLastApplied(2);
 275         abstractBehavior.performSnapshotWithoutCapture(1);
 276         assertEquals(1, abstractBehavior.getReplicatedToAllIndex());
 277         assertEquals(1, context.getReplicatedLog().size());
 278     }
 279
 280
 281     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(final MockRaftActorContext actorContext,
 282             final ActorRef actorRef, final RaftRPC rpc) {
 283
 284         Payload payload = new MockRaftActorContext.MockPayload("");
 285         setLastLogEntry(actorContext, 1, 0, payload);
 286         actorContext.getTermInformation().update(1, "test");
 287
 288         RaftActorBehavior origBehavior = createBehavior(actorContext);
 289         RaftActorBehavior raftBehavior = origBehavior.handleMessage(actorRef, rpc);
 290
 291         assertEquals("New raft state", RaftState.Follower, raftBehavior.state());
 292         assertEquals("New election term", rpc.getTerm(), actorContext.getTermInformation().getCurrentTerm());
 293
 294         origBehavior.close();
 295         raftBehavior.close();
 296     }
 297
 298     protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry(
 299         final MockRaftActorContext actorContext, final long term, final long index, final Payload data) {
 300         return setLastLogEntry(actorContext, new SimpleReplicatedLogEntry(index, term, data));
 301     }
 302
 303     protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry(final MockRaftActorContext actorContext,
 304             final ReplicatedLogEntry logEntry) {
 305         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 306         log.append(logEntry);
 307         actorContext.setReplicatedLog(log);
 308
 309         return log;
 310     }
 311
 312     protected abstract T createBehavior(RaftActorContext actorContext);
 313
 314     protected final T createBehavior(final MockRaftActorContext actorContext) {
 315         T ret = createBehavior((RaftActorContext)actorContext);
 316         actorContext.setCurrentBehavior(ret);
 317         return ret;
 318     }
 319
 320     protected RaftActorBehavior createBehavior() {
 321         return createBehavior(createActorContext());
 322     }
 323
 324     protected MockRaftActorContext createActorContext() {
 325         return new MockRaftActorContext();
 326     }
 327
 328     protected MockRaftActorContext createActorContext(final ActorRef actor) {
 329         return new MockRaftActorContext("test", getSystem(), actor);
 330     }
 331
 332     protected AppendEntries createAppendEntriesWithNewerTerm() {
 333         return new AppendEntries(100, "leader-1", 0, 0, Collections.emptyList(), 1, -1, (short)0);
 334     }
 335
 336     protected AppendEntriesReply createAppendEntriesReplyWithNewerTerm() {
 337         return new AppendEntriesReply("follower-1", 100, false, 100, 100, (short)0);
 338     }
 339
 340     protected RequestVote createRequestVoteWithNewerTerm() {
 341         return new RequestVote(100, "candidate-1", 10, 100);
 342     }
 343
 344     protected RequestVoteReply createRequestVoteReplyWithNewerTerm() {
 345         return new RequestVoteReply(100, false);
 346     }
 347
 348     protected ByteString toByteString(final Map<String, String> state) {
 349         ByteArrayOutputStream bos = new ByteArrayOutputStream();
 350         try (ObjectOutputStream oos = new ObjectOutputStream(bos)) {
 351             oos.writeObject(state);
 352             return ByteString.copyFrom(bos.toByteArray());
 353         } catch (IOException e) {
 354             throw new AssertionError("IOException occurred converting Map to Bytestring", e);
 355         }
 356     }
 357
 358     protected void logStart(final String name) {
 359         LoggerFactory.getLogger(getClass()).info("Starting " + name);
 360     }
 361
 362     protected RaftPolicy createRaftPolicy(final boolean automaticElectionsEnabled,
 363                                           final boolean applyModificationToStateBeforeConsensus) {
 364         return new RaftPolicy() {
 365             @Override
 366             public boolean automaticElectionsEnabled() {
 367                 return automaticElectionsEnabled;
 368             }
 369
 370             @Override
 371             public boolean applyModificationToStateBeforeConsensus() {
 372                 return applyModificationToStateBeforeConsensus;
 373             }
 374         };
 375     }
 376 }