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%2FRaftActorSnapshotMessageSupport.java;h=bc96713dc46f31f7974ddb02a70ee50a3f1f62b7;hb=d0f46920468c8e4b67c68bd9058572b2d10d75f1;hp=b35f4babb1a4999452a2703229a7eebf48a57fb2;hpb=ad860a3b51c31b740aabb297727e15aa45756777;p=controller.git 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 b35f4babb1..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 @@ -10,17 +10,21 @@ package org.opendaylight.controller.cluster.raft; import akka.actor.ActorRef; import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; +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.apache.commons.lang3.SerializationUtils; 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.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.Duration; +import scala.concurrent.duration.FiniteDuration; /** * Handles snapshot related messages for a RaftActor. @@ -39,30 +43,37 @@ class RaftActorSnapshotMessageSupport { private final RaftActorSnapshotCohort cohort; private final Logger log; - private Duration snapshotReplyActorTimeout = Duration.create(30, TimeUnit.SECONDS); + private FiniteDuration snapshotReplyActorTimeout = FiniteDuration.create(30, TimeUnit.SECONDS); RaftActorSnapshotMessageSupport(final RaftActorContext context, final RaftActorSnapshotCohort cohort) { this.context = context; this.cohort = cohort; - this.log = context.getLogger(); + log = context.getLogger(); - context.getSnapshotManager().setCreateSnapshotRunnable(() -> cohort.createSnapshot(context.getActor())); - context.getSnapshotManager().setApplySnapshotConsumer(cohort::applySnapshot); + context.getSnapshotManager().setCreateSnapshotConsumer( + outputStream -> cohort.createSnapshot(context.getActor(), outputStream)); + context.getSnapshotManager().setSnapshotCohort(cohort); } - boolean handleSnapshotMessage(Object message, ActorRef sender) { - if (message instanceof ApplySnapshot ) { + 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); } else if (message instanceof SaveSnapshotFailure) { onSaveSnapshotFailure((SaveSnapshotFailure) message); } else if (message instanceof CaptureSnapshotReply) { - onCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot()); + onCaptureSnapshotReply((CaptureSnapshotReply) message); } else if (COMMIT_SNAPSHOT.equals(message)) { - context.getSnapshotManager().commit(-1); + context.getSnapshotManager().commit(-1, -1); } else if (message instanceof GetSnapshot) { - onGetSnapshot(sender); + onGetSnapshot(sender, (GetSnapshot) message); + } else if (message instanceof SnapshotComplete) { + log.debug("{}: SnapshotComplete received", context.getId()); } else { return false; } @@ -70,57 +81,63 @@ class RaftActorSnapshotMessageSupport { return true; } - private void onCaptureSnapshotReply(byte[] snapshotBytes) { - log.debug("{}: CaptureSnapshotReply received by actor: snapshot size {}", context.getId(), snapshotBytes.length); + private void onCaptureSnapshotReply(final CaptureSnapshotReply reply) { + log.debug("{}: CaptureSnapshotReply received by actor", context.getId()); - context.getSnapshotManager().persist(snapshotBytes, context.getTotalMemory()); + 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(sequenceNumber); + log.info("{}: SaveSnapshotSuccess received for snapshot, sequenceNr: {}", context.getId(), sequenceNumber); + + context.getSnapshotManager().commit(sequenceNumber, success.metadata().timestamp()); } - private void onApplySnapshot(ApplySnapshot message) { + private void onApplySnapshot(final ApplySnapshot message) { log.info("{}: Applying snapshot on follower: {}", context.getId(), message.getSnapshot()); context.getSnapshotManager().apply(message); } - private void onGetSnapshot(ActorRef sender) { + private void onGetSnapshot(final ActorRef sender, final GetSnapshot getSnapshot) { log.debug("{}: onGetSnapshot", context.getId()); - if(context.getPersistenceProvider().isRecoveryApplicable()) { + + if (context.getPersistenceProvider().isRecoveryApplicable()) { CaptureSnapshot captureSnapshot = context.getSnapshotManager().newCaptureSnapshot( - context.getReplicatedLog().last(), -1, false); + context.getReplicatedLog().last(), -1, true); + + final FiniteDuration timeout = + getSnapshot.getTimeout().map(Timeout::duration).orElse(snapshotReplyActorTimeout); ActorRef snapshotReplyActor = context.actorOf(GetSnapshotReplyActor.props(captureSnapshot, - ImmutableElectionTerm.copyOf(context.getTermInformation()), sender, - snapshotReplyActorTimeout, context.getId(), context.getPeerServerInfo(true))); + ImmutableElectionTerm.copyOf(context.getTermInformation()), sender, timeout, context.getId(), + context.getPeerServerInfo(true))); - cohort.createSnapshot(snapshotReplyActor); + cohort.createSnapshot(snapshotReplyActor, Optional.empty()); } else { - Snapshot snapshot = Snapshot.create(new byte[0], Collections.emptyList(), -1, -1, -1, -1, + 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(), SerializationUtils.serialize(snapshot)), - context.getActor()); + sender.tell(new GetSnapshotReply(context.getId(), snapshot), context.getActor()); } } @VisibleForTesting - void setSnapshotReplyActorTimeout(Duration snapshotReplyActorTimeout) { + void setSnapshotReplyActorTimeout(final FiniteDuration snapshotReplyActorTimeout) { this.snapshotReplyActorTimeout = snapshotReplyActorTimeout; } }