X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FAbstractLeader.java;h=6560ad76c3937285300173f16df53f4c60d9ae1d;hb=HEAD;hp=32e2e09c3fe2b13482a4f02a2a460e5ef45643dd;hpb=8da08c9f008d371c491cf6b35494ca36c2ebdcb9;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java index 32e2e09c3f..7fdb73dc0f 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractLeader.java @@ -18,9 +18,7 @@ import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.io.IOException; import java.io.ObjectOutputStream; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; -import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -34,19 +32,21 @@ import org.opendaylight.controller.cluster.io.SharedFileBackedOutputStream; import org.opendaylight.controller.cluster.messaging.MessageSlicer; import org.opendaylight.controller.cluster.messaging.SliceOptions; import org.opendaylight.controller.cluster.raft.ClientRequestTracker; -import org.opendaylight.controller.cluster.raft.ClientRequestTrackerImpl; import org.opendaylight.controller.cluster.raft.FollowerLogInformation; import org.opendaylight.controller.cluster.raft.PeerInfo; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; +import org.opendaylight.controller.cluster.raft.RaftVersions; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; import org.opendaylight.controller.cluster.raft.VotingState; +import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; import org.opendaylight.controller.cluster.raft.base.messages.CheckConsensusReached; import org.opendaylight.controller.cluster.raft.base.messages.Replicate; import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat; import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; +import org.opendaylight.controller.cluster.raft.messages.IdentifiablePayload; import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot; import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply; import org.opendaylight.controller.cluster.raft.messages.RaftRPC; @@ -107,7 +107,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { super(context, state); appendEntriesMessageSlicer = MessageSlicer.builder().logContext(logName()) - .messageSliceSize(context.getConfigParams().getSnapshotChunkSize()) + .messageSliceSize(context.getConfigParams().getMaximumMessageSliceSize()) .expireStateAfterInactivity(context.getConfigParams().getElectionTimeOutInterval().toMillis() * 3, TimeUnit.MILLISECONDS).build(); @@ -163,7 +163,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerToLog.remove(followerId); } - public void updateMinReplicaCount() { + public final void updateMinReplicaCount() { int numVoting = 0; for (PeerInfo peer: context.getPeers()) { if (peer.isVoting()) { @@ -218,6 +218,13 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { return this; } + final var followerRaftVersion = appendEntriesReply.getRaftVersion(); + if (followerRaftVersion < RaftVersions.FLUORINE_VERSION) { + log.warn("{}: handleAppendEntriesReply - ignoring reply from follower {} raft version {}", logName(), + followerId, followerRaftVersion); + return this; + } + final long lastActivityNanos = followerLogInformation.nanosSinceLastActivity(); if (lastActivityNanos > context.getConfigParams().getElectionTimeOutInterval().toNanos()) { log.warn("{} : handleAppendEntriesReply delayed beyond election timeout, " @@ -228,7 +235,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerLogInformation.markFollowerActive(); followerLogInformation.setPayloadVersion(appendEntriesReply.getPayloadVersion()); - followerLogInformation.setRaftVersion(appendEntriesReply.getRaftVersion()); + followerLogInformation.setRaftVersion(followerRaftVersion); followerLogInformation.setNeedsLeaderAddress(appendEntriesReply.isNeedsLeaderAddress()); long followerLastLogIndex = appendEntriesReply.getLogLastIndex(); @@ -379,13 +386,13 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // However we keep looping so we can make progress when new entries in the current term // reach consensus, as per §5.4.1: "once an entry from the current term is committed by // counting replicas, then all prior entries are committed indirectly". - if (replicatedLogEntry.getTerm() == currentTerm()) { + if (replicatedLogEntry.term() == currentTerm()) { log.trace("{}: Setting commit index to {}", logName(), index); context.setCommitIndex(index); } else { log.debug("{}: Not updating commit index to {} - retrieved log entry with index {}, " + "term {} does not match the current term {}", logName(), index, - replicatedLogEntry.getIndex(), replicatedLogEntry.getTerm(), currentTerm()); + replicatedLogEntry.index(), replicatedLogEntry.term(), currentTerm()); } } else { log.trace("{}: minReplicationCount not reached, actual {} - breaking", logName(), replicatedCount); @@ -433,20 +440,44 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { super.performSnapshotWithoutCapture(minReplicatedToAllIndex); } - @Override - protected ClientRequestTracker removeClientRequestTracker(final long logIndex) { - final Iterator it = trackers.iterator(); + /** + * Removes and returns the ClientRequestTracker for the specified log index. + * @param logIndex the log index + * @return the ClientRequestTracker or null if none available + */ + private ClientRequestTracker removeClientRequestTracker(final long logIndex) { + final var it = trackers.iterator(); while (it.hasNext()) { - final ClientRequestTracker t = it.next(); - if (t.getIndex() == logIndex) { + final var tracker = it.next(); + if (tracker.logIndex() == logIndex) { it.remove(); - return t; + return tracker; } } - return null; } + @Override + final ApplyState getApplyStateFor(final ReplicatedLogEntry entry) { + // first check whether a ClientRequestTracker exists for this entry. + // If it does that means the leader wasn't dropped before the transaction applied. + // That means that this transaction can be safely applied as a local transaction since we + // have the ClientRequestTracker. + final var tracker = removeClientRequestTracker(entry.index()); + if (tracker != null) { + return new ApplyState(tracker.clientActor(), tracker.identifier(), entry); + } + + // Tracker is missing, this means that we switched behaviours between replicate and applystate + // and became the leader again,. We still want to apply this as a local modification because + // we have resumed leadership with that log entry having been committed. + if (entry.getData() instanceof IdentifiablePayload identifiable) { + return new ApplyState(null, identifiable.getIdentifier(), entry); + } + + return new ApplyState(null, null, entry); + } + @Override protected RaftActorBehavior handleRequestVoteReply(final ActorRef sender, final RequestVoteReply requestVoteReply) { return this; @@ -464,47 +495,45 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { return this; } - if (message instanceof RaftRPC) { - RaftRPC rpc = (RaftRPC) message; - // If RPC request or response contains term T > currentTerm: - // set currentTerm = T, convert to follower (§5.1) - // This applies to all RPC messages and responses - if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) { - log.info("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower", - logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); - - context.getTermInformation().updateAndPersist(rpc.getTerm(), null); - - // This is a special case. Normally when stepping down as leader we don't process and reply to the - // RaftRPC as per raft. But if we're in the process of transferring leadership and we get a - // RequestVote, process the RequestVote before switching to Follower. This enables the requesting - // candidate node to be elected the leader faster and avoids us possibly timing out in the Follower - // state and starting a new election and grabbing leadership back before the other candidate node can - // start a new election due to lack of responses. This case would only occur if there isn't a majority - // of other nodes available that can elect the requesting candidate. Since we're transferring - // leadership, we should make every effort to get the requesting node elected. - if (rpc instanceof RequestVote && context.getRaftActorLeadershipTransferCohort() != null) { - log.debug("{}: Leadership transfer in progress - processing RequestVote", logName()); - super.handleMessage(sender, rpc); - } - - return internalSwitchBehavior(RaftState.Follower); + // If RPC request or response contains term T > currentTerm: + // set currentTerm = T, convert to follower (§5.1) + // This applies to all RPC messages and responses + if (message instanceof RaftRPC rpc && rpc.getTerm() > context.getTermInformation().getCurrentTerm() + && shouldUpdateTerm(rpc)) { + + log.info("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower", + logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); + + context.getTermInformation().updateAndPersist(rpc.getTerm(), null); + + // This is a special case. Normally when stepping down as leader we don't process and reply to the + // RaftRPC as per raft. But if we're in the process of transferring leadership and we get a + // RequestVote, process the RequestVote before switching to Follower. This enables the requesting + // candidate node to be elected the leader faster and avoids us possibly timing out in the Follower + // state and starting a new election and grabbing leadership back before the other candidate node can + // start a new election due to lack of responses. This case would only occur if there isn't a majority + // of other nodes available that can elect the requesting candidate. Since we're transferring + // leadership, we should make every effort to get the requesting node elected. + if (rpc instanceof RequestVote requestVote && context.getRaftActorLeadershipTransferCohort() != null) { + log.debug("{}: Leadership transfer in progress - processing RequestVote", logName()); + requestVote(sender, requestVote); } + + return internalSwitchBehavior(RaftState.Follower); } if (message instanceof SendHeartBeat) { beforeSendHeartbeat(); sendHeartBeat(); scheduleHeartBeat(context.getConfigParams().getHeartBeatInterval()); - } else if (message instanceof SendInstallSnapshot) { - SendInstallSnapshot sendInstallSnapshot = (SendInstallSnapshot) message; + } else if (message instanceof SendInstallSnapshot sendInstallSnapshot) { setSnapshotHolder(new SnapshotHolder(sendInstallSnapshot.getSnapshot(), sendInstallSnapshot.getSnapshotBytes())); sendInstallSnapshot(); - } else if (message instanceof Replicate) { - replicate((Replicate) message); - } else if (message instanceof InstallSnapshotReply) { - handleInstallSnapshotReply((InstallSnapshotReply) message); + } else if (message instanceof Replicate replicate) { + replicate(replicate); + } else if (message instanceof InstallSnapshotReply installSnapshotReply) { + handleInstallSnapshotReply(installSnapshotReply); } else if (message instanceof CheckConsensusReached) { possiblyUpdateCommitIndex(); } else { @@ -544,7 +573,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if (installSnapshotState.isLastChunk(reply.getChunkIndex())) { //this was the last chunk reply - long followerMatchIndex = snapshotHolder.get().getLastIncludedIndex(); + long followerMatchIndex = snapshotHolder.orElseThrow().getLastIncludedIndex(); followerLogInformation.setMatchIndex(followerMatchIndex); followerLogInformation.setNextIndex(followerMatchIndex + 1); followerLogInformation.clearLeaderInstallSnapshotState(); @@ -612,17 +641,16 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } private void replicate(final Replicate replicate) { - long logIndex = replicate.getReplicatedLogEntry().getIndex(); + final long logIndex = replicate.logIndex(); - log.debug("{}: Replicate message: identifier: {}, logIndex: {}, payload: {}, isSendImmediate: {}", logName(), - replicate.getIdentifier(), logIndex, replicate.getReplicatedLogEntry().getData().getClass(), - replicate.isSendImmediate()); + log.debug("{}: Replicate message: identifier: {}, logIndex: {}, isSendImmediate: {}", logName(), + replicate.identifier(), logIndex, replicate.sendImmediate()); // Create a tracker entry we will use this later to notify the // client actor - if (replicate.getClientActor() != null) { - trackers.add(new ClientRequestTrackerImpl(replicate.getClientActor(), replicate.getIdentifier(), - logIndex)); + final var clientActor = replicate.clientActor(); + if (clientActor != null) { + trackers.add(new ClientRequestTracker(logIndex, clientActor, replicate.identifier())); } boolean applyModificationToState = !context.anyVotingPeers() @@ -633,7 +661,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { applyLogToStateMachine(logIndex); } - if (replicate.isSendImmediate() && !followerToLog.isEmpty()) { + if (replicate.sendImmediate() && !followerToLog.isEmpty()) { sendAppendEntries(0, false); } } @@ -663,7 +691,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { long followerNextIndex = followerLogInformation.getNextIndex(); boolean isFollowerActive = followerLogInformation.isFollowerActive(); boolean sendAppendEntries = false; - List entries = Collections.emptyList(); + var entries = List.of(); LeaderInstallSnapshotState installSnapshotState = followerLogInformation.getInstallSnapshotState(); if (installSnapshotState != null) { @@ -747,24 +775,24 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { // Try to get all the entries in the journal but not exceeding the max data size for a single AppendEntries // message. int maxEntries = (int) context.getReplicatedLog().size(); - final int maxDataSize = context.getConfigParams().getSnapshotChunkSize(); + final int maxDataSize = context.getConfigParams().getMaximumMessageSliceSize(); final long followerNextIndex = followerLogInfo.getNextIndex(); - List entries = context.getReplicatedLog().getFrom(followerNextIndex, - maxEntries, maxDataSize); + final var entries = context.getReplicatedLog().getFrom(followerNextIndex, maxEntries, maxDataSize); // If the first entry's size exceeds the max data size threshold, it will be returned from the call above. If // that is the case, then we need to slice it into smaller chunks. - if (!(entries.size() == 1 && entries.get(0).getData().size() > maxDataSize)) { + if (entries.size() != 1 || entries.get(0).getData().serializedSize() <= maxDataSize) { // Don't need to slice. return entries; } - log.debug("{}: Log entry size {} exceeds max payload size {}", logName(), entries.get(0).getData().size(), + final var firstEntry = entries.get(0); + log.debug("{}: Log entry size {} exceeds max payload size {}", logName(), firstEntry.getData().size(), maxDataSize); // If an AppendEntries has already been serialized for the log index then reuse the // SharedFileBackedOutputStream. - final Long logIndex = entries.get(0).getIndex(); + final Long logIndex = firstEntry.index(); SharedFileBackedOutputStream fileBackedStream = sharedSerializedAppendEntriesStreams.get(logIndex); if (fileBackedStream == null) { fileBackedStream = context.getFileBackedOutputStreamFactory().newSharedInstance(); @@ -781,7 +809,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } catch (IOException e) { log.error("{}: Error serializing {}", logName(), appendEntries, e); fileBackedStream.cleanup(); - return Collections.emptyList(); + return List.of(); } sharedSerializedAppendEntriesStreams.put(logIndex, fileBackedStream); @@ -809,7 +837,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { followerLogInfo.setSlicedLogEntryIndex(FollowerLogInformation.NO_INDEX); }).build()); - return Collections.emptyList(); + return List.of(); } private void sendAppendEntriesToFollower(final ActorSelection followerActor, final List entries, @@ -874,11 +902,11 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { return true; } - boolean captureInitiated = context.getSnapshotManager().captureToInstall(context.getReplicatedLog().last(), - this.getReplicatedToAllIndex(), followerId); + boolean captureInitiated = context.getSnapshotManager().captureToInstall(context.getReplicatedLog().lastMeta(), + getReplicatedToAllIndex(), followerId); if (captureInitiated) { followerLogInfo.setLeaderInstallSnapshotState(new LeaderInstallSnapshotState( - context.getConfigParams().getSnapshotChunkSize(), logName())); + context.getConfigParams().getMaximumMessageSliceSize(), logName())); } return captureInitiated; @@ -920,14 +948,14 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { if (snapshotHolder.isPresent()) { LeaderInstallSnapshotState installSnapshotState = followerLogInfo.getInstallSnapshotState(); if (installSnapshotState == null) { - installSnapshotState = new LeaderInstallSnapshotState(context.getConfigParams().getSnapshotChunkSize(), - logName()); + installSnapshotState = new LeaderInstallSnapshotState( + context.getConfigParams().getMaximumMessageSliceSize(), logName()); followerLogInfo.setLeaderInstallSnapshotState(installSnapshotState); } try { // Ensure the snapshot bytes are set - this is a no-op. - installSnapshotState.setSnapshotBytes(snapshotHolder.get().getSnapshotBytes()); + installSnapshotState.setSnapshotBytes(snapshotHolder.orElseThrow().getSnapshotBytes()); if (!installSnapshotState.canSendNextChunk()) { return; @@ -952,7 +980,7 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { } catch (IOException e) { log.warn("{}: Unable to send chunk: {}/{}. Reseting snapshot progress. Snapshot state: {}", logName(), installSnapshotState.getChunkIndex(), installSnapshotState.getTotalChunks(), - installSnapshotState); + installSnapshotState, e); installSnapshotState.reset(); } } @@ -966,14 +994,14 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { installSnapshotState.startChunkTimer(); followerActor.tell( new InstallSnapshot(currentTerm(), context.getId(), - snapshotHolder.get().getLastIncludedIndex(), - snapshotHolder.get().getLastIncludedTerm(), + snapshotHolder.orElseThrow().getLastIncludedIndex(), + snapshotHolder.orElseThrow().getLastIncludedTerm(), snapshotChunk, chunkIndex, installSnapshotState.getTotalChunks(), OptionalInt.of(installSnapshotState.getLastChunkHashCode()), - serverConfig - ).toSerializable(followerLogInfo.getRaftVersion()), + serverConfig, + followerLogInfo.getRaftVersion()), actor() ); } @@ -1095,8 +1123,8 @@ public abstract class AbstractLeader extends AbstractRaftActorBehavior { private final ByteSource snapshotBytes; SnapshotHolder(final Snapshot snapshot, final ByteSource snapshotBytes) { - this.lastIncludedTerm = snapshot.getLastAppliedTerm(); - this.lastIncludedIndex = snapshot.getLastAppliedIndex(); + lastIncludedTerm = snapshot.getLastAppliedTerm(); + lastIncludedIndex = snapshot.getLastAppliedIndex(); this.snapshotBytes = snapshotBytes; }