X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActorSnapshotMessageSupport.java;h=bc96713dc46f31f7974ddb02a70ee50a3f1f62b7;hp=21c8ffa68e44763a14f0d5865ad03d8104e25d60;hb=e131c3498d286ff14890120ff5e9020ba89f10f9;hpb=c31a6fcf9fb070d4419ca4c32d8b531fdcb5030d diff --git a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java index 21c8ffa68e..bc96713dc4 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java +++ b/opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/RaftActorSnapshotMessageSupport.java @@ -8,15 +8,23 @@ package org.opendaylight.controller.cluster.raft; import akka.actor.ActorRef; -import akka.japi.Procedure; import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; -import org.opendaylight.controller.cluster.DataPersistenceProvider; +import akka.util.Timeout; +import com.google.common.annotations.VisibleForTesting; +import java.util.Collections; +import java.util.Optional; +import java.util.concurrent.TimeUnit; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; -import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; +import org.opendaylight.controller.cluster.raft.base.messages.SnapshotComplete; +import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot; +import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshotReply; +import org.opendaylight.controller.cluster.raft.persisted.EmptyState; +import org.opendaylight.controller.cluster.raft.persisted.Snapshot; import org.slf4j.Logger; +import scala.concurrent.duration.FiniteDuration; /** * Handles snapshot related messages for a RaftActor. @@ -24,95 +32,112 @@ import org.slf4j.Logger; * @author Thomas Pantelis */ class RaftActorSnapshotMessageSupport { - static final String COMMIT_SNAPSHOT = "commit_snapshot"; + static final Object COMMIT_SNAPSHOT = new Object() { + @Override + public String toString() { + return "commit_snapshot"; + } + }; - private final DataPersistenceProvider persistence; private final RaftActorContext context; - private final RaftActorBehavior currentBehavior; private final RaftActorSnapshotCohort cohort; - private final ActorRef raftActorRef; private final Logger log; - private final Procedure createSnapshotProcedure = new Procedure() { - @Override - public void apply(Void notUsed) throws Exception { - cohort.createSnapshot(raftActorRef); - } - }; + private FiniteDuration snapshotReplyActorTimeout = FiniteDuration.create(30, TimeUnit.SECONDS); - RaftActorSnapshotMessageSupport(DataPersistenceProvider persistence, RaftActorContext context, - RaftActorBehavior currentBehavior, RaftActorSnapshotCohort cohort, ActorRef raftActorRef) { - this.persistence = persistence; + RaftActorSnapshotMessageSupport(final RaftActorContext context, final RaftActorSnapshotCohort cohort) { this.context = context; - this.currentBehavior = currentBehavior; this.cohort = cohort; - this.raftActorRef = raftActorRef; - this.log = context.getLogger(); + log = context.getLogger(); + + context.getSnapshotManager().setCreateSnapshotConsumer( + outputStream -> cohort.createSnapshot(context.getActor(), outputStream)); + context.getSnapshotManager().setSnapshotCohort(cohort); } - boolean handleSnapshotMessage(Object message) { - if(message instanceof ApplySnapshot ) { - onApplySnapshot(((ApplySnapshot) message).getSnapshot()); - return true; + RaftActorSnapshotCohort getSnapshotCohort() { + return cohort; + } + + boolean handleSnapshotMessage(final Object message, final ActorRef sender) { + if (message instanceof ApplySnapshot) { + onApplySnapshot((ApplySnapshot) message); } else if (message instanceof SaveSnapshotSuccess) { onSaveSnapshotSuccess((SaveSnapshotSuccess) message); - return true; } else if (message instanceof SaveSnapshotFailure) { onSaveSnapshotFailure((SaveSnapshotFailure) message); - return true; - } else if (message instanceof CaptureSnapshot) { - onCaptureSnapshot(message); - return true; } else if (message instanceof CaptureSnapshotReply) { - onCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot()); - return true; - } else if (message.equals(COMMIT_SNAPSHOT)) { - context.getSnapshotManager().commit(persistence, -1); - return true; + onCaptureSnapshotReply((CaptureSnapshotReply) message); + } else if (COMMIT_SNAPSHOT.equals(message)) { + context.getSnapshotManager().commit(-1, -1); + } else if (message instanceof GetSnapshot) { + onGetSnapshot(sender, (GetSnapshot) message); + } else if (message instanceof SnapshotComplete) { + log.debug("{}: SnapshotComplete received", context.getId()); } else { return false; } - } - private void onCaptureSnapshotReply(byte[] snapshotBytes) { - log.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", context.getId(), snapshotBytes.length); - - context.getSnapshotManager().persist(persistence, snapshotBytes, currentBehavior, context.getTotalMemory()); + return true; } - private void onCaptureSnapshot(Object message) { - log.debug("{}: CaptureSnapshot received by actor: {}", context.getId(), message); + private void onCaptureSnapshotReply(final CaptureSnapshotReply reply) { + log.debug("{}: CaptureSnapshotReply received by actor", context.getId()); - context.getSnapshotManager().create(createSnapshotProcedure); + context.getSnapshotManager().persist(reply.getSnapshotState(), reply.getInstallSnapshotStream(), + context.getTotalMemory()); } - private void onSaveSnapshotFailure(SaveSnapshotFailure saveSnapshotFailure) { + private void onSaveSnapshotFailure(final SaveSnapshotFailure saveSnapshotFailure) { log.error("{}: SaveSnapshotFailure received for snapshot Cause:", context.getId(), saveSnapshotFailure.cause()); context.getSnapshotManager().rollback(); } - private void onSaveSnapshotSuccess(SaveSnapshotSuccess success) { - log.info("{}: SaveSnapshotSuccess received for snapshot", context.getId()); - + private void onSaveSnapshotSuccess(final SaveSnapshotSuccess success) { long sequenceNumber = success.metadata().sequenceNr(); - context.getSnapshotManager().commit(persistence, sequenceNumber); + log.info("{}: SaveSnapshotSuccess received for snapshot, sequenceNr: {}", context.getId(), sequenceNumber); + + context.getSnapshotManager().commit(sequenceNumber, success.metadata().timestamp()); } - private void onApplySnapshot(Snapshot snapshot) { - if(log.isDebugEnabled()) { - log.debug("{}: ApplySnapshot called on Follower Actor " + - "snapshotIndex:{}, snapshotTerm:{}", context.getId(), snapshot.getLastAppliedIndex(), - snapshot.getLastAppliedTerm()); - } + private void onApplySnapshot(final ApplySnapshot message) { + log.info("{}: Applying snapshot on follower: {}", context.getId(), message.getSnapshot()); + + context.getSnapshotManager().apply(message); + } + + private void onGetSnapshot(final ActorRef sender, final GetSnapshot getSnapshot) { + log.debug("{}: onGetSnapshot", context.getId()); + + + if (context.getPersistenceProvider().isRecoveryApplicable()) { + CaptureSnapshot captureSnapshot = context.getSnapshotManager().newCaptureSnapshot( + context.getReplicatedLog().last(), -1, true); - cohort.applySnapshot(snapshot.getState()); + final FiniteDuration timeout = + getSnapshot.getTimeout().map(Timeout::duration).orElse(snapshotReplyActorTimeout); + + ActorRef snapshotReplyActor = context.actorOf(GetSnapshotReplyActor.props(captureSnapshot, + ImmutableElectionTerm.copyOf(context.getTermInformation()), sender, timeout, context.getId(), + context.getPeerServerInfo(true))); + + cohort.createSnapshot(snapshotReplyActor, Optional.empty()); + } else { + Snapshot snapshot = Snapshot.create( + EmptyState.INSTANCE, Collections.emptyList(), + -1, -1, -1, -1, + context.getTermInformation().getCurrentTerm(), context.getTermInformation().getVotedFor(), + context.getPeerServerInfo(true)); + + sender.tell(new GetSnapshotReply(context.getId(), snapshot), context.getActor()); + } + } - //clears the followers log, sets the snapshot index to ensure adjusted-index works - context.setReplicatedLog(ReplicatedLogImpl.newInstance(snapshot, context, persistence, - currentBehavior)); - context.setLastApplied(snapshot.getLastAppliedIndex()); + @VisibleForTesting + void setSnapshotReplyActorTimeout(final FiniteDuration snapshotReplyActorTimeout) { + this.snapshotReplyActorTimeout = snapshotReplyActorTimeout; } }