2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore.admin;
10 import akka.actor.ActorRef;
11 import akka.actor.Status.Success;
12 import akka.dispatch.OnComplete;
13 import akka.pattern.Patterns;
14 import akka.util.Timeout;
16 import com.google.common.base.Function;
17 import com.google.common.base.Strings;
18 import com.google.common.base.Throwables;
19 import com.google.common.util.concurrent.FutureCallback;
20 import com.google.common.util.concurrent.Futures;
21 import com.google.common.util.concurrent.ListenableFuture;
22 import com.google.common.util.concurrent.MoreExecutors;
23 import com.google.common.util.concurrent.SettableFuture;
24 import java.io.FileOutputStream;
25 import java.io.IOException;
26 import java.util.AbstractMap.SimpleEntry;
27 import java.util.ArrayList;
28 import java.util.HashMap;
29 import java.util.List;
31 import java.util.Map.Entry;
33 import java.util.concurrent.Future;
34 import java.util.concurrent.TimeUnit;
35 import org.apache.commons.lang3.SerializationUtils;
36 import org.opendaylight.controller.cluster.access.concepts.MemberName;
37 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
38 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
39 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
40 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
41 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
42 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
43 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
44 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
45 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
46 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
47 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
48 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
49 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
50 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
51 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
52 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
81 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
82 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
83 import org.opendaylight.yangtools.yang.common.RpcResult;
84 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
85 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
86 import org.slf4j.Logger;
87 import org.slf4j.LoggerFactory;
90 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
92 * @author Thomas Pantelis
94 public class ClusterAdminRpcService implements ClusterAdminService {
95 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
97 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
99 private final DistributedDataStoreInterface configDataStore;
100 private final DistributedDataStoreInterface operDataStore;
101 private final BindingNormalizedNodeSerializer serializer;
102 private final Timeout makeLeaderLocalTimeout;
104 public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
105 DistributedDataStoreInterface operDataStore,
106 BindingNormalizedNodeSerializer serializer) {
107 this.configDataStore = configDataStore;
108 this.operDataStore = operDataStore;
109 this.serializer = serializer;
111 this.makeLeaderLocalTimeout =
112 new Timeout(configDataStore.getActorContext().getDatastoreContext()
113 .getShardLeaderElectionTimeout().duration().$times(2));
117 public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
118 final String shardName = input.getShardName();
119 if (Strings.isNullOrEmpty(shardName)) {
120 return newFailedRpcResultFuture("A valid shard name must be specified");
123 DataStoreType dataStoreType = input.getDataStoreType();
124 if (dataStoreType == null) {
125 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
128 LOG.info("Adding replica for shard {}", shardName);
130 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
131 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
132 Futures.addCallback(future, new FutureCallback<Success>() {
134 public void onSuccess(Success success) {
135 LOG.info("Successfully added replica for shard {}", shardName);
136 returnFuture.set(newSuccessfulResult());
140 public void onFailure(Throwable failure) {
141 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
142 returnFuture, failure);
144 }, MoreExecutors.directExecutor());
150 public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
151 final String shardName = input.getShardName();
152 if (Strings.isNullOrEmpty(shardName)) {
153 return newFailedRpcResultFuture("A valid shard name must be specified");
156 DataStoreType dataStoreType = input.getDataStoreType();
157 if (dataStoreType == null) {
158 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
161 final String memberName = input.getMemberName();
162 if (Strings.isNullOrEmpty(memberName)) {
163 return newFailedRpcResultFuture("A valid member name must be specified");
166 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
168 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
169 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
170 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
171 Futures.addCallback(future, new FutureCallback<Success>() {
173 public void onSuccess(Success success) {
174 LOG.info("Successfully removed replica for shard {}", shardName);
175 returnFuture.set(newSuccessfulResult());
179 public void onFailure(Throwable failure) {
180 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
181 returnFuture, failure);
183 }, MoreExecutors.directExecutor());
189 public Future<RpcResult<Void>> makeLeaderLocal(final MakeLeaderLocalInput input) {
190 final String shardName = input.getShardName();
191 if (Strings.isNullOrEmpty(shardName)) {
192 return newFailedRpcResultFuture("A valid shard name must be specified");
195 DataStoreType dataStoreType = input.getDataStoreType();
196 if (dataStoreType == null) {
197 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
200 ActorContext actorContext = dataStoreType == DataStoreType.Config
201 ? configDataStore.getActorContext()
202 : operDataStore.getActorContext();
204 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
205 actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
207 final scala.concurrent.Future<ActorRef> localShardReply =
208 actorContext.findLocalShardAsync(shardName);
210 final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
211 localShardReply.onComplete(new OnComplete<ActorRef>() {
213 public void onComplete(final Throwable failure, final ActorRef actorRef) throws Throwable {
214 if (failure != null) {
215 LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
216 + " local shard.", shardName, failure);
217 makeLeaderLocalAsk.failure(failure);
220 .completeWith(actorContext
221 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
224 }, actorContext.getClientDispatcher());
226 final SettableFuture<RpcResult<Void>> future = SettableFuture.create();
227 makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
229 public void onComplete(final Throwable failure, final Object success) throws Throwable {
230 if (failure != null) {
231 LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
232 future.set(RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION,
233 "leadership transfer failed", failure).build());
237 LOG.debug("Leadership transfer complete");
238 future.set(RpcResultBuilder.<Void>success().build());
240 }, actorContext.getClientDispatcher());
246 public Future<RpcResult<Void>> addPrefixShardReplica(final AddPrefixShardReplicaInput input) {
248 final InstanceIdentifier<?> identifier = input.getShardPrefix();
249 if (identifier == null) {
250 return newFailedRpcResultFuture("A valid shard identifier must be specified");
253 final DataStoreType dataStoreType = input.getDataStoreType();
254 if (dataStoreType == null) {
255 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
258 LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
260 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
261 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
262 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
263 Futures.addCallback(future, new FutureCallback<Success>() {
265 public void onSuccess(Success success) {
266 LOG.info("Successfully added replica for shard {}", prefix);
267 returnFuture.set(newSuccessfulResult());
271 public void onFailure(Throwable failure) {
272 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
273 returnFuture, failure);
275 }, MoreExecutors.directExecutor());
281 public Future<RpcResult<Void>> removePrefixShardReplica(final RemovePrefixShardReplicaInput input) {
283 final InstanceIdentifier<?> identifier = input.getShardPrefix();
284 if (identifier == null) {
285 return newFailedRpcResultFuture("A valid shard identifier must be specified");
288 final DataStoreType dataStoreType = input.getDataStoreType();
289 if (dataStoreType == null) {
290 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
293 final String memberName = input.getMemberName();
294 if (Strings.isNullOrEmpty(memberName)) {
295 return newFailedRpcResultFuture("A valid member name must be specified");
298 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
299 identifier, memberName, dataStoreType);
300 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
302 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
303 final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
304 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
305 Futures.addCallback(future, new FutureCallback<Success>() {
307 public void onSuccess(final Success success) {
308 LOG.info("Successfully removed replica for shard {}", prefix);
309 returnFuture.set(newSuccessfulResult());
313 public void onFailure(final Throwable failure) {
314 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
315 returnFuture, failure);
317 }, MoreExecutors.directExecutor());
323 public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
324 LOG.info("Adding replicas for all shards");
326 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
327 Function<String, Object> messageSupplier = AddShardReplica::new;
329 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
330 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
332 return waitForShardResults(shardResultData, shardResults ->
333 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
334 "Failed to add replica");
339 public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
340 LOG.info("Removing replicas for all shards");
342 final String memberName = input.getMemberName();
343 if (Strings.isNullOrEmpty(memberName)) {
344 return newFailedRpcResultFuture("A valid member name must be specified");
347 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
348 Function<String, Object> messageSupplier = shardName ->
349 new RemoveShardReplica(shardName, MemberName.forName(memberName));
351 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
352 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
354 return waitForShardResults(shardResultData, shardResults ->
355 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
356 " Failed to remove replica");
360 public Future<RpcResult<Void>> changeMemberVotingStatesForShard(ChangeMemberVotingStatesForShardInput input) {
361 final String shardName = input.getShardName();
362 if (Strings.isNullOrEmpty(shardName)) {
363 return newFailedRpcResultFuture("A valid shard name must be specified");
366 DataStoreType dataStoreType = input.getDataStoreType();
367 if (dataStoreType == null) {
368 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
371 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
372 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
373 return newFailedRpcResultFuture("No member voting state input was specified");
376 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
379 LOG.info("Change member voting states for shard {}: {}", shardName,
380 changeVotingStatus.getMeberVotingStatusMap());
382 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
383 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
384 Futures.addCallback(future, new FutureCallback<Success>() {
386 public void onSuccess(Success success) {
387 LOG.info("Successfully changed member voting states for shard {}", shardName);
388 returnFuture.set(newSuccessfulResult());
392 public void onFailure(Throwable failure) {
393 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
394 returnFuture, failure);
396 }, MoreExecutors.directExecutor());
402 public Future<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
403 final ChangeMemberVotingStatesForAllShardsInput input) {
404 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
405 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
406 return newFailedRpcResultFuture("No member voting state input was specified");
409 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
410 Function<String, Object> messageSupplier = shardName ->
411 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
413 LOG.info("Change member voting states for all shards");
415 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
416 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
418 return waitForShardResults(shardResultData, shardResults ->
419 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
420 "Failed to change member voting states");
424 public Future<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards() {
425 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
426 Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
428 LOG.info("Flip member voting states for all shards");
430 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
431 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
433 return waitForShardResults(shardResultData, shardResults ->
434 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
435 "Failed to change member voting states");
439 public Future<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
440 final String shardName = input.getShardName();
441 if (Strings.isNullOrEmpty(shardName)) {
442 return newFailedRpcResultFuture("A valid shard name must be specified");
445 DataStoreType dataStoreType = input.getDataStoreType();
446 if (dataStoreType == null) {
447 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
450 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
452 final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
453 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
454 new GetShardRole(shardName));
455 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
457 public void onSuccess(final GetShardRoleReply reply) {
459 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
460 "No Shard role present. Please retry..").build());
463 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
464 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
465 if (reply.getRole() != null) {
466 builder.setRole(reply.getRole());
468 returnFuture.set(newSuccessfulResult(builder.build()));
472 public void onFailure(final Throwable failure) {
473 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
474 "Failed to get shard role.", failure).build());
476 }, MoreExecutors.directExecutor());
482 public Future<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(final GetPrefixShardRoleInput input) {
483 final InstanceIdentifier<?> identifier = input.getShardPrefix();
484 if (identifier == null) {
485 return newFailedRpcResultFuture("A valid shard identifier must be specified");
488 final DataStoreType dataStoreType = input.getDataStoreType();
489 if (dataStoreType == null) {
490 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
493 LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
495 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
496 final String shardName = ClusterUtils.getCleanShardName(prefix);
497 final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
498 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
499 new GetShardRole(shardName));
500 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
502 public void onSuccess(final GetShardRoleReply reply) {
504 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
505 "No Shard role present. Please retry..").build());
509 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
510 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
511 if (reply.getRole() != null) {
512 builder.setRole(reply.getRole());
514 returnFuture.set(newSuccessfulResult(builder.build()));
518 public void onFailure(final Throwable failure) {
519 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
520 "Failed to get shard role.", failure).build());
522 }, MoreExecutors.directExecutor());
528 public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
529 LOG.debug("backupDatastore: {}", input);
531 if (Strings.isNullOrEmpty(input.getFilePath())) {
532 return newFailedRpcResultFuture("A valid file path must be specified");
535 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
536 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
537 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
539 public void onSuccess(List<DatastoreSnapshot> snapshots) {
540 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
544 public void onFailure(Throwable failure) {
545 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
547 }, MoreExecutors.directExecutor());
552 private ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
553 List<MemberVotingState> memberVotingStatus) {
554 Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
555 for (MemberVotingState memberStatus: memberVotingStatus) {
556 serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
558 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
561 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
562 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
563 final Function<List<ShardResult>, T> resultDataSupplier,
564 final String failureLogMsgPrefix) {
565 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
566 final List<ShardResult> shardResults = new ArrayList<>();
567 for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
568 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
570 public void onSuccess(Success result) {
571 synchronized (shardResults) {
572 ShardResultBuilder shardResult = entry.getValue();
573 LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
574 shardResult.getDataStoreType());
575 shardResults.add(shardResult.setSucceeded(true).build());
581 public void onFailure(Throwable failure) {
582 synchronized (shardResults) {
583 ShardResultBuilder shardResult = entry.getValue();
584 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
585 shardResult.getDataStoreType(), failure);
586 shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
587 Throwables.getRootCause(failure).getMessage()).build());
592 void checkIfComplete() {
593 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
594 if (shardResults.size() == shardResultData.size()) {
595 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
598 }, MoreExecutors.directExecutor());
603 private <T> void sendMessageToManagerForConfiguredShards(DataStoreType dataStoreType,
604 List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
605 Function<String, Object> messageSupplier) {
606 ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
607 : operDataStore.getActorContext();
608 Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
610 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
612 for (String shardName: allShardNames) {
613 ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
615 shardResultData.add(new SimpleEntry<>(future,
616 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
620 @SuppressWarnings("unchecked")
621 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
622 Timeout timeout = SHARD_MGR_TIMEOUT;
623 ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
624 ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
626 return Futures.allAsList(configFuture, operFuture);
629 private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
630 ActorRef shardManager = dataStoreType == DataStoreType.Config
631 ? configDataStore.getActorContext().getShardManager()
632 : operDataStore.getActorContext().getShardManager();
633 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
636 @SuppressWarnings("checkstyle:IllegalCatch")
637 private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
638 SettableFuture<RpcResult<Void>> returnFuture) {
639 try (FileOutputStream fos = new FileOutputStream(fileName)) {
640 SerializationUtils.serialize(snapshots, fos);
642 returnFuture.set(newSuccessfulResult());
643 LOG.info("Successfully backed up datastore to file {}", fileName);
644 } catch (IOException | RuntimeException e) {
645 onDatastoreBackupFailure(fileName, returnFuture, e);
649 private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
651 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
654 private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
656 LOG.error(msg, failure);
657 returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
658 failure.getMessage())).build());
661 private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
662 final SettableFuture<T> returnFuture = SettableFuture.create();
664 @SuppressWarnings("unchecked")
665 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
666 askFuture.onComplete(new OnComplete<T>() {
668 public void onComplete(Throwable failure, T resp) {
669 if (failure != null) {
670 returnFuture.setException(failure);
672 returnFuture.set(resp);
675 }, configDataStore.getActorContext().getClientDispatcher());
680 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
681 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
684 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
685 return newFailedRpcResultBuilder(message, null);
688 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
689 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
692 private static RpcResult<Void> newSuccessfulResult() {
693 return newSuccessfulResult(null);
696 private static <T> RpcResult<T> newSuccessfulResult(T data) {
697 return RpcResultBuilder.success(data).build();