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

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