X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FAbstractRaftActorBehavior.java;h=8cb011f7a71d1b54b76bba86a261612f4f555bb0;hb=refs%2Fchanges%2F18%2F28718%2F6;hp=dbeafe9eb8b2fce467451eb4594c6c1be913797e;hpb=77d55c2a5a0311aac06707d71e199ba30271b48c;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java index dbeafe9eb8..8cb011f7a7 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehavior.java @@ -10,20 +10,23 @@ package org.opendaylight.controller.cluster.raft.behaviors; import akka.actor.ActorRef; import akka.actor.Cancellable; -import akka.event.LoggingAdapter; +import java.util.Map; import java.util.Random; import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.raft.ClientRequestTracker; 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.base.messages.ApplyLogEntries; +import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload; +import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.messages.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; +import org.slf4j.Logger; import scala.concurrent.duration.FiniteDuration; /** @@ -38,6 +41,8 @@ import scala.concurrent.duration.FiniteDuration; */ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { + protected static final ElectionTimeout ELECTION_TIMEOUT = new ElectionTimeout(); + /** * Information about the RaftActor whose behavior this class represents */ @@ -46,7 +51,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** * */ - protected final LoggingAdapter LOG; + protected final Logger LOG; /** * @@ -58,10 +63,39 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { */ protected String leaderId = null; + private short leaderPayloadVersion = -1; + + private long replicatedToAllIndex = -1; + + private final String logName; + + private final RaftState state; - protected AbstractRaftActorBehavior(RaftActorContext context) { + protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) { this.context = context; + this.state = state; this.LOG = context.getLogger(); + + logName = String.format("%s (%s)", context.getId(), state); + } + + @Override + public RaftState state() { + return state; + } + + public String logName() { + return logName; + } + + @Override + public void setReplicatedToAllIndex(long replicatedToAllIndex) { + this.replicatedToAllIndex = replicatedToAllIndex; + } + + @Override + public long getReplicatedToAllIndex() { + return replicatedToAllIndex; } /** @@ -74,7 +108,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntries The AppendEntries message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleAppendEntries(ActorRef sender, AppendEntries appendEntries); @@ -86,7 +120,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender * @param appendEntries - * @return + * @return a new behavior if it was changed or the current behavior */ protected RaftActorBehavior appendEntries(ActorRef sender, AppendEntries appendEntries) { @@ -95,12 +129,12 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { if (appendEntries.getTerm() < currentTerm()) { if(LOG.isDebugEnabled()) { LOG.debug("{}: Cannot append entries because sender term {} is less than {}", - context.getId(), appendEntries.getTerm(), currentTerm()); + logName(), appendEntries.getTerm(), currentTerm()); } sender.tell( new AppendEntriesReply(context.getId(), currentTerm(), false, - lastIndex(), lastTerm()), actor() + lastIndex(), lastTerm(), context.getPayloadVersion()), actor() ); return this; } @@ -119,7 +153,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param appendEntriesReply The AppendEntriesReply message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleAppendEntriesReply(ActorRef sender, AppendEntriesReply appendEntriesReply); @@ -130,15 +164,28 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender * @param requestVote - * @return + * @return a new behavior if it was changed or the current behavior */ - protected RaftActorBehavior requestVote(ActorRef sender, - RequestVote requestVote) { + protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) { - if(LOG.isDebugEnabled()) { - LOG.debug("{}: Received {}", context.getId(), requestVote); + LOG.debug("{}: In requestVote: {}", logName(), requestVote); + + boolean grantVote = canGrantVote(requestVote); + + if(grantVote) { + context.getTermInformation().updateAndPersist(requestVote.getTerm(), requestVote.getCandidateId()); } + RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote); + + LOG.debug("{}: requestVote returning: {}", logName(), reply); + + sender.tell(reply, actor()); + + return this; + } + + protected boolean canGrantVote(RequestVote requestVote){ boolean grantVote = false; // Reply false if term < currentTerm (§5.1) @@ -148,7 +195,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // If votedFor is null or candidateId, and candidate’s log is at // least as up-to-date as receiver’s log, grant vote (§5.2, §5.4) } else if (votedFor() == null || votedFor() - .equals(requestVote.getCandidateId())) { + .equals(requestVote.getCandidateId())) { boolean candidateLatest = false; @@ -162,20 +209,15 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { if (requestVote.getLastLogTerm() > lastTerm()) { candidateLatest = true; } else if ((requestVote.getLastLogTerm() == lastTerm()) - && requestVote.getLastLogIndex() >= lastIndex()) { + && requestVote.getLastLogIndex() >= lastIndex()) { candidateLatest = true; } if (candidateLatest) { grantVote = true; - context.getTermInformation().updateAndPersist(requestVote.getTerm(), - requestVote.getCandidateId()); } } - - sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor()); - - return this; + return grantVote; } /** @@ -188,15 +230,14 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { * * @param sender The actor that sent this message * @param requestVoteReply The RequestVoteReply message - * @return + * @return a new behavior if it was changed or the current behavior */ protected abstract RaftActorBehavior handleRequestVoteReply(ActorRef sender, RequestVoteReply requestVoteReply); /** - * Creates a random election duration * - * @return + * @return a random election duration */ protected FiniteDuration electionDuration() { long variance = new Random().nextInt(context.getConfigParams().getElectionTimeVariance()); @@ -216,7 +257,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** * schedule a new election * - * @param interval + * @param interval the duration after which we should trigger a new election */ protected void scheduleElection(FiniteDuration interval) { stopElection(); @@ -225,67 +266,57 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // message is sent to itself electionCancel = context.getActorSystem().scheduler().scheduleOnce(interval, - context.getActor(), new ElectionTimeout(), + context.getActor(), ELECTION_TIMEOUT, context.getActorSystem().dispatcher(), context.getActor()); } /** - * Get the current term - * @return + * @return the current term */ protected long currentTerm() { return context.getTermInformation().getCurrentTerm(); } /** - * Get the candidate for whom we voted in the current term - * @return + * @return the candidate for whom we voted in the current term */ protected String votedFor() { return context.getTermInformation().getVotedFor(); } /** - * Get the actor associated with this behavior - * @return + * @return the actor associated with this behavior */ protected ActorRef actor() { return context.getActor(); } /** - * Get the term from the last entry in the log * - * @return + * @return the term from the last entry in the log */ protected long lastTerm() { return context.getReplicatedLog().lastTerm(); } /** - * Get the index from the last entry in the log - * - * @return + * @return the index from the last entry in the log */ protected long lastIndex() { return context.getReplicatedLog().lastIndex(); } /** - * Find the client request tracker for a specific logIndex - * * @param logIndex - * @return + * @return the client request tracker for the specified logIndex */ protected ClientRequestTracker findClientRequestTracker(long logIndex) { return null; } /** - * Find the client request tracker for a specific logIndex - * * @param logIndex - * @return + * @return the client request tracker for the specified logIndex */ protected ClientRequestTracker removeClientRequestTracker(long logIndex) { return null; @@ -293,9 +324,8 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { /** - * Find the log index from the previous to last entry in the log * - * @return + * @return log index from the previous to last entry in the log */ protected long prevLogIndex(long index){ ReplicatedLogEntry prevEntry = @@ -307,8 +337,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } /** - * Find the log term from the previous to last entry in the log - * @return + * @return log term from the previous to last entry in the log */ protected long prevLogTerm(long index){ ReplicatedLogEntry prevEntry = @@ -349,14 +378,14 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { } else { //if one index is not present in the log, no point in looping // around as the rest wont be present either - LOG.warning( + LOG.warn( "{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}", - context.getId(), i, i, index); + logName(), i, i, index); break; } } if(LOG.isDebugEnabled()) { - LOG.debug("{}: Setting last applied to {}", context.getId(), newLastApplied); + LOG.debug("{}: Setting last applied to {}", logName(), newLastApplied); } context.setLastApplied(newLastApplied); @@ -364,7 +393,7 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { // will be used during recovery //in case if the above code throws an error and this message is not sent, it would be fine // as the append entries received later would initiate add this message to the journal - actor().tell(new ApplyLogEntries((int) context.getLastApplied()), actor()); + actor().tell(new ApplyJournalEntries(context.getLastApplied()), actor()); } protected Object fromSerializableMessage(Object serializable){ @@ -389,17 +418,38 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return leaderId; } - protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { - LOG.info("{} :- Switching from behavior {} to {}", context.getId(), this.state(), behavior.state()); + @Override + public short getLeaderPayloadVersion() { + return leaderPayloadVersion; + } + + public void setLeaderPayloadVersion(short leaderPayloadVersion) { + this.leaderPayloadVersion = leaderPayloadVersion; + } + + @Override + public RaftActorBehavior switchBehavior(RaftActorBehavior behavior) { + return internalSwitchBehavior(behavior); + } + + protected RaftActorBehavior internalSwitchBehavior(RaftState newState) { + if(context.getRaftPolicy().automaticElectionsEnabled()){ + return internalSwitchBehavior(newState.createBehavior(context)); + } + return this; + } + + private RaftActorBehavior internalSwitchBehavior(RaftActorBehavior newBehavior) { + LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), newBehavior.state()); try { close(); } catch (Exception e) { - LOG.error(e, "{}: Failed to close behavior : {}", context.getId(), this.state()); + LOG.error("{}: Failed to close behavior : {}", logName(), this.state(), e); } - - return behavior; + return newBehavior; } + protected int getMajorityVoteCount(int numPeers) { // Votes are required from a majority of the peers including self. // The numMajority field therefore stores a calculated value @@ -422,4 +472,40 @@ public abstract class AbstractRaftActorBehavior implements RaftActorBehavior { return numMajority; } + + + /** + * Performs a snapshot with no capture on the replicated log. + * It clears the log from the supplied index or last-applied-1 which ever is minimum. + * + * @param snapshotCapturedIndex + */ + protected void performSnapshotWithoutCapture(final long snapshotCapturedIndex) { + long actualIndex = context.getSnapshotManager().trimLog(snapshotCapturedIndex, this); + + if(actualIndex != -1){ + setReplicatedToAllIndex(actualIndex); + } + } + + protected String getId(){ + return context.getId(); + } + + public void applyServerConfiguration(ServerConfigurationPayload serverConfig) { + Map currentPeers = context.getPeerAddresses(); + for(String peerId: serverConfig.getNewServerConfig()) { + if(!getId().equals(peerId)) { + if(!currentPeers.containsKey(peerId)) { + context.addToPeers(peerId, null); + } else { + currentPeers.remove(peerId); + } + } + } + + for(String peerIdToRemove: currentPeers.keySet()) { + context.removePeer(peerIdToRemove); + } + } }