Add optional timeout parameter for backup rpc
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / main / java / org / opendaylight / controller / cluster / raft / RaftActorSnapshotMessageSupport.java
1 /*
2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.controller.cluster.raft;
9
10 import akka.actor.ActorRef;
11 import akka.persistence.SaveSnapshotFailure;
12 import akka.persistence.SaveSnapshotSuccess;
13 import akka.util.Timeout;
14 import com.google.common.annotations.VisibleForTesting;
15 import java.util.Collections;
16 import java.util.Optional;
17 import java.util.concurrent.TimeUnit;
18 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
19 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
20 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
21 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
22 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshotReply;
23 import org.opendaylight.controller.cluster.raft.persisted.EmptyState;
24 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
25 import org.slf4j.Logger;
26 import scala.concurrent.duration.FiniteDuration;
27
28 /**
29  * Handles snapshot related messages for a RaftActor.
30  *
31  * @author Thomas Pantelis
32  */
33 class RaftActorSnapshotMessageSupport {
34     static final Object COMMIT_SNAPSHOT = new Object() {
35         @Override
36         public String toString() {
37             return "commit_snapshot";
38         }
39     };
40
41     private final RaftActorContext context;
42     private final RaftActorSnapshotCohort cohort;
43     private final Logger log;
44
45     private FiniteDuration snapshotReplyActorTimeout = FiniteDuration.create(30, TimeUnit.SECONDS);
46
47     RaftActorSnapshotMessageSupport(final RaftActorContext context, final RaftActorSnapshotCohort cohort) {
48         this.context = context;
49         this.cohort = cohort;
50         this.log = context.getLogger();
51
52         context.getSnapshotManager().setCreateSnapshotConsumer(
53             outputStream -> cohort.createSnapshot(context.getActor(), outputStream));
54         context.getSnapshotManager().setSnapshotCohort(cohort);
55     }
56
57     RaftActorSnapshotCohort getSnapshotCohort() {
58         return cohort;
59     }
60
61     boolean handleSnapshotMessage(Object message, ActorRef sender) {
62         if (message instanceof ApplySnapshot) {
63             onApplySnapshot((ApplySnapshot) message);
64         } else if (message instanceof SaveSnapshotSuccess) {
65             onSaveSnapshotSuccess((SaveSnapshotSuccess) message);
66         } else if (message instanceof SaveSnapshotFailure) {
67             onSaveSnapshotFailure((SaveSnapshotFailure) message);
68         } else if (message instanceof CaptureSnapshotReply) {
69             onCaptureSnapshotReply((CaptureSnapshotReply) message);
70         } else if (COMMIT_SNAPSHOT.equals(message)) {
71             context.getSnapshotManager().commit(-1, -1);
72         } else if (message instanceof GetSnapshot) {
73             onGetSnapshot(sender, (GetSnapshot) message);
74         } else {
75             return false;
76         }
77
78         return true;
79     }
80
81     private void onCaptureSnapshotReply(CaptureSnapshotReply reply) {
82         log.debug("{}: CaptureSnapshotReply received by actor", context.getId());
83
84         context.getSnapshotManager().persist(reply.getSnapshotState(), reply.getInstallSnapshotStream(),
85                 context.getTotalMemory());
86     }
87
88     private void onSaveSnapshotFailure(SaveSnapshotFailure saveSnapshotFailure) {
89         log.error("{}: SaveSnapshotFailure received for snapshot Cause:",
90                 context.getId(), saveSnapshotFailure.cause());
91
92         context.getSnapshotManager().rollback();
93     }
94
95     private void onSaveSnapshotSuccess(SaveSnapshotSuccess success) {
96         long sequenceNumber = success.metadata().sequenceNr();
97
98         log.info("{}: SaveSnapshotSuccess received for snapshot, sequenceNr: {}", context.getId(), sequenceNumber);
99
100         context.getSnapshotManager().commit(sequenceNumber, success.metadata().timestamp());
101     }
102
103     private void onApplySnapshot(ApplySnapshot message) {
104         log.info("{}: Applying snapshot on follower:  {}", context.getId(), message.getSnapshot());
105
106         context.getSnapshotManager().apply(message);
107     }
108
109     private void onGetSnapshot(ActorRef sender, GetSnapshot getSnapshot) {
110         log.debug("{}: onGetSnapshot", context.getId());
111
112
113         if (context.getPersistenceProvider().isRecoveryApplicable()) {
114             CaptureSnapshot captureSnapshot = context.getSnapshotManager().newCaptureSnapshot(
115                     context.getReplicatedLog().last(), -1, true);
116
117             final FiniteDuration timeout =
118                     getSnapshot.getTimeout().map(Timeout::duration).orElse(snapshotReplyActorTimeout);
119
120             ActorRef snapshotReplyActor = context.actorOf(GetSnapshotReplyActor.props(captureSnapshot,
121                     ImmutableElectionTerm.copyOf(context.getTermInformation()), sender, timeout, context.getId(),
122                     context.getPeerServerInfo(true)));
123
124             cohort.createSnapshot(snapshotReplyActor, Optional.empty());
125         } else {
126             Snapshot snapshot = Snapshot.create(
127                     EmptyState.INSTANCE, Collections.<ReplicatedLogEntry>emptyList(),
128                     -1, -1, -1, -1,
129                     context.getTermInformation().getCurrentTerm(), context.getTermInformation().getVotedFor(),
130                     context.getPeerServerInfo(true));
131
132             sender.tell(new GetSnapshotReply(context.getId(), snapshot), context.getActor());
133         }
134     }
135
136     @VisibleForTesting
137     void setSnapshotReplyActorTimeout(FiniteDuration snapshotReplyActorTimeout) {
138         this.snapshotReplyActorTimeout = snapshotReplyActorTimeout;
139     }
140 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.