X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FAbstractRaftActorBehaviorTest.java;h=d3f5a0eead1ff3ccdd2e69f215eccfb6886143b4;hb=655216a6c75aa29d31c4c56c56a5000db56ba233;hp=1e7aa6a1d5fe40e2fd57e3088e80dc0dcbe1d653;hpb=7be62e955c32ff7fa10753c4307199b287b1904c;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java index 1e7aa6a1d5..d3f5a0eead 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java @@ -1,213 +1,352 @@ package org.opendaylight.controller.cluster.raft.behaviors; +import static org.junit.Assert.assertEquals; import akka.actor.ActorRef; import akka.actor.Props; -import akka.testkit.JavaTestKit; +import akka.testkit.TestActorRef; +import com.google.protobuf.ByteString; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import org.junit.After; +import org.junit.Assert; import org.junit.Test; import org.opendaylight.controller.cluster.raft.AbstractActorTest; import org.opendaylight.controller.cluster.raft.MockRaftActorContext; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; +import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.SerializationUtils; +import org.opendaylight.controller.cluster.raft.TestActorFactory; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.messages.RaftRPC; import org.opendaylight.controller.cluster.raft.messages.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; -import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; +import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; +import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; +import org.slf4j.LoggerFactory; -import java.util.concurrent.atomic.AtomicLong; +public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest { -import static org.junit.Assert.assertEquals; + protected final TestActorFactory actorFactory = new TestActorFactory(getSystem()); -public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest{ + private final TestActorRef behaviorActor = actorFactory.createTestActor( + Props.create(MessageCollectorActor.class), actorFactory.generateActorId("behavior")); - private final ActorRef behaviorActor = getSystem().actorOf(Props.create( - DoNothingActor.class)); + RaftActorBehavior behavior; - @Test - public void testHandlingOfRaftRPCWithNewerTerm() throws Exception { - new JavaTestKit(getSystem()) {{ + @After + public void tearDown() throws Exception { + if(behavior != null) { + behavior.close(); + } - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + actorFactory.close(); + } + + /** + * This test checks that when a new Raft RPC message is received with a newer + * term the RaftActor gets into the Follower state. + * + * @throws Exception + */ + @Test + public void testHandleRaftRPCWithNewerTerm() throws Exception { + RaftActorContext actorContext = createActorContext(); + + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createAppendEntriesWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createAppendEntriesReplyWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createRequestVoteWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createRequestVoteReplyWithNewerTerm()); - - - }}; } + + /** + * This test verifies that when an AppendEntries is received with a term that + * is less that the currentTerm of the RaftActor then the RaftActor does not + * change it's state and it responds back with a failure + * + * @throws Exception + */ @Test - public void testHandlingOfAppendEntriesWithNewerCommitIndex() throws Exception{ - new JavaTestKit(getSystem()) {{ + public void testHandleAppendEntriesSenderTermLessThanReceiverTerm() throws Exception { + MockRaftActorContext context = createActorContext(); + short payloadVersion = 5; + context.setPayloadVersion(payloadVersion); - RaftActorContext context = - createActorContext(); + // First set the receivers term to a high number (1000) + context.getTermInformation().update(1000, "test"); - ((MockRaftActorContext) context).setLastApplied(100); + AppendEntries appendEntries = new AppendEntries(100, "leader-1", 0, 0, null, 101, -1, (short)4); - AppendEntries appendEntries = - new AppendEntries(100, "leader-1", 0, 0, null, 101); + behavior = createBehavior(context); - RaftState raftState = - createBehavior(context).handleMessage(getRef(), appendEntries); + // Send an unknown message so that the state of the RaftActor remains unchanged + RaftActorBehavior expected = behavior.handleMessage(behaviorActor, "unknown"); - assertEquals(new AtomicLong(101).get(), context.getLastApplied()); + RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries); - }}; + assertEquals("Raft state", expected.state(), raftBehavior.state()); + + // Also expect an AppendEntriesReply to be sent where success is false + + AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching( + behaviorActor, AppendEntriesReply.class); + + assertEquals("isSuccess", false, reply.isSuccess()); + assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion()); } + @Test - public void testHandleRequestVoteWhenSenderTermGreaterThanCurrentTermAndSenderLogMoreUpToDate(){ - new JavaTestKit(getSystem()) {{ + public void testHandleAppendEntriesAddSameEntryToLog() throws Exception { + MockRaftActorContext context = createActorContext(); + + context.getTermInformation().update(2, "test"); + + // Prepare the receivers log + MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("zero"); + setLastLogEntry(context, 2, 0, payload); + + List entries = new ArrayList<>(); + entries.add(new MockRaftActorContext.MockReplicatedLogEntry(2, 0, payload)); + + AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 2, -1, (short)0); + + behavior = createBehavior(context); - new Within(duration("1 seconds")) { - protected void run() { + if (behavior instanceof Candidate) { + // Resetting the Candidates term to make sure it will match + // the term sent by AppendEntries. If this was not done then + // the test will fail because the Candidate will assume that + // the message was sent to it from a lower term peer and will + // thus respond with a failure + context.getTermInformation().update(2, "test"); + } - RaftActorBehavior follower = createBehavior( - createActorContext(behaviorActor)); + // Send an unknown message so that the state of the RaftActor remains unchanged + RaftActorBehavior expected = behavior.handleMessage(behaviorActor, "unknown"); - follower.handleMessage(getTestActor(), new RequestVote(1000, "test", 10000, 999)); + RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries); - final Boolean out = new ExpectMsg(duration("1 seconds"), "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); + assertEquals("Raft state", expected.state(), raftBehavior.state()); - assertEquals(true, out); - } - }; - }}; + assertEquals("ReplicatedLog size", 1, context.getReplicatedLog().size()); + + handleAppendEntriesAddSameEntryToLogReply(behaviorActor); + } + + protected void handleAppendEntriesAddSameEntryToLogReply(TestActorRef replyActor) + throws Exception { + AppendEntriesReply reply = MessageCollectorActor.getFirstMatching(replyActor, AppendEntriesReply.class); + Assert.assertNull("Expected no AppendEntriesReply", reply); } + /** + * This test verifies that when a RequestVote is received by the RaftActor + * with the senders' log is more up to date than the receiver that the receiver grants + * the vote to the sender. + */ @Test - public void testHandleRequestVoteWhenSenderTermGreaterThanCurrentTermButSenderLogLessUptoDate(){ - new JavaTestKit(getSystem()) {{ + public void testHandleRequestVoteWhenSenderLogMoreUpToDate() { + MockRaftActorContext context = createActorContext(); - new Within(duration("1 seconds")) { - protected void run() { + behavior = createBehavior(context); - RaftActorContext actorContext = - createActorContext(behaviorActor); + context.getTermInformation().update(1, "test"); - MockRaftActorContext.MockReplicatedLog log = new MockRaftActorContext.MockReplicatedLog(); - log.setReplicatedLogEntry(new MockRaftActorContext.MockReplicatedLogEntry(20000, 1000000, "")); - log.setLast( - new MockRaftActorContext.MockReplicatedLogEntry(20000, - 1000000, "")); + behavior.handleMessage(behaviorActor, new RequestVote(context.getTermInformation().getCurrentTerm(), + "test", 10000, 999)); - ((MockRaftActorContext) actorContext).setReplicatedLog(log); + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", true, reply.isVoteGranted()); + } - RaftActorBehavior follower = createBehavior(actorContext); + /** + * This test verifies that when a RaftActor receives a RequestVote message + * with a term that is greater than it's currentTerm but a less up-to-date + * log then the receiving RaftActor will not grant the vote to the sender + */ + @Test + public void testHandleRequestVoteWhenSenderLogLessUptoDate() { + MockRaftActorContext context = createActorContext(); - follower.handleMessage(getTestActor(), new RequestVote(1000, "test", 10000, 999)); + behavior = createBehavior(context); - final Boolean out = new ExpectMsg(duration("1 seconds"), "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); + context.getTermInformation().update(1, "test"); - assertEquals(false, out); - } - }; - }}; + int index = 2000; + setLastLogEntry(context, context.getTermInformation().getCurrentTerm(), index, + new MockRaftActorContext.MockPayload("")); + + behavior.handleMessage(behaviorActor, new RequestVote( + context.getTermInformation().getCurrentTerm(), "test", + index - 1, context.getTermInformation().getCurrentTerm())); + + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", false, reply.isVoteGranted()); } + /** + * This test verifies that the receiving RaftActor will not grant a vote + * to a sender if the sender's term is lesser than the currentTerm of the + * recipient RaftActor + */ + @Test + public void testHandleRequestVoteWhenSenderTermLessThanCurrentTerm() { + RaftActorContext context = createActorContext(); + + context.getTermInformation().update(1000, null); + + behavior = createBehavior(context); + + behavior.handleMessage(behaviorActor, new RequestVote(999, "test", 10000, 999)); + + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", false, reply.isVoteGranted()); + } @Test - public void testHandleRequestVoteWhenSenderTermLessThanCurrentTerm(){ - new JavaTestKit(getSystem()) {{ + public void testPerformSnapshot() { + MockRaftActorContext context = new MockRaftActorContext("test", getSystem(), behaviorActor); + AbstractRaftActorBehavior abstractBehavior = (AbstractRaftActorBehavior) createBehavior(context); + if (abstractBehavior instanceof Candidate) { + return; + } + + context.getTermInformation().update(1, "test"); + + //log has 1 entry with replicatedToAllIndex = 0, does not do anything, returns the + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build()); + context.setLastApplied(0); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(-1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); + + //2 entries, lastApplied still 0, no purging. + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build()); + context.setLastApplied(0); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(-1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(2, context.getReplicatedLog().size()); + + //2 entries, lastApplied still 0, no purging. + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build()); + context.setLastApplied(1); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(0, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); + + //5 entries, lastApplied =2 and replicatedIndex = 3, but since we want to keep the lastapplied, indices 0 and 1 will only get purged + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 5, 1).build()); + context.setLastApplied(2); + abstractBehavior.performSnapshotWithoutCapture(3); + assertEquals(1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(3, context.getReplicatedLog().size()); + + // scenario where Last applied > Replicated to all index (becoz of a slow follower) + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build()); + context.setLastApplied(2); + abstractBehavior.performSnapshotWithoutCapture(1); + assertEquals(1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); + } - new Within(duration("1 seconds")) { - protected void run() { - RaftActorContext context = createActorContext(behaviorActor); + protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext, + ActorRef actorRef, RaftRPC rpc) throws Exception { - context.getTermInformation().update(1000, null); + Payload p = new MockRaftActorContext.MockPayload(""); + setLastLogEntry((MockRaftActorContext) actorContext, 1, 0, p); + actorContext.getTermInformation().update(1, "test"); - RaftActorBehavior follower = createBehavior(context); + RaftActorBehavior origBehavior = createBehavior(actorContext); + RaftActorBehavior raftBehavior = origBehavior.handleMessage(actorRef, rpc); - follower.handleMessage(getTestActor(), new RequestVote(999, "test", 10000, 999)); + assertEquals("New raft state", RaftState.Follower, raftBehavior.state()); + assertEquals("New election term", rpc.getTerm(), actorContext.getTermInformation().getCurrentTerm()); - final Boolean out = new ExpectMsg(duration("1 seconds"), "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); + origBehavior.close(); + raftBehavior.close(); + } - assertEquals(false, out); - } - }; - }}; + protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry( + MockRaftActorContext actorContext, long term, long index, Payload data) { + return setLastLogEntry(actorContext, + new MockRaftActorContext.MockReplicatedLogEntry(term, index, data)); } - protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm( - ActorRef actorRef, RaftRPC rpc){ - RaftState raftState = createBehavior() - .handleMessage(actorRef, rpc); + protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry(MockRaftActorContext actorContext, + ReplicatedLogEntry logEntry) { + MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); + log.append(logEntry); + actorContext.setReplicatedLog(log); - assertEquals(RaftState.Follower, raftState); + return log; } - protected abstract RaftActorBehavior createBehavior(RaftActorContext actorContext); + protected abstract RaftActorBehavior createBehavior( + RaftActorContext actorContext); - protected RaftActorBehavior createBehavior(){ + protected RaftActorBehavior createBehavior() { return createBehavior(createActorContext()); } - protected RaftActorContext createActorContext(){ + protected MockRaftActorContext createActorContext() { return new MockRaftActorContext(); } - protected RaftActorContext createActorContext(ActorRef actor) { + protected MockRaftActorContext createActorContext(ActorRef actor) { return new MockRaftActorContext("test", getSystem(), actor); } - protected AppendEntries createAppendEntriesWithNewerTerm(){ - return new AppendEntries(100, "leader-1", 0, 0, null, 1); + protected AppendEntries createAppendEntriesWithNewerTerm() { + return new AppendEntries(100, "leader-1", 0, 0, null, 1, -1, (short)0); } - protected AppendEntriesReply createAppendEntriesReplyWithNewerTerm(){ - return new AppendEntriesReply(100, false); + protected AppendEntriesReply createAppendEntriesReplyWithNewerTerm() { + return new AppendEntriesReply("follower-1", 100, false, 100, 100, (short)0); } - protected RequestVote createRequestVoteWithNewerTerm(){ + protected RequestVote createRequestVoteWithNewerTerm() { return new RequestVote(100, "candidate-1", 10, 100); } - protected RequestVoteReply createRequestVoteReplyWithNewerTerm(){ + protected RequestVoteReply createRequestVoteReplyWithNewerTerm() { return new RequestVoteReply(100, false); } + protected Object fromSerializableMessage(Object serializable){ + return SerializationUtils.fromSerializable(serializable); + } + protected ByteString toByteString(Map state) { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try(ObjectOutputStream oos = new ObjectOutputStream(bos)) { + oos.writeObject(state); + return ByteString.copyFrom(bos.toByteArray()); + } catch (IOException e) { + throw new AssertionError("IOException occurred converting Map to Bytestring", e); + } + } + protected void logStart(String name) { + LoggerFactory.getLogger(LeaderTest.class).info("Starting " + name); + } }