opendaylight/md-sal/sal-cluster-admin-impl/src/main/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcService.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.datastore.admin;
   9
  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;
  15 import com.google.common.base.Function;
  16 import com.google.common.base.Strings;
  17 import com.google.common.base.Throwables;
  18 import com.google.common.util.concurrent.FutureCallback;
  19 import com.google.common.util.concurrent.Futures;
  20 import com.google.common.util.concurrent.ListenableFuture;
  21 import com.google.common.util.concurrent.SettableFuture;
  22 import java.io.FileOutputStream;
  23 import java.io.IOException;
  24 import java.util.AbstractMap.SimpleEntry;
  25 import java.util.ArrayList;
  26 import java.util.HashMap;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.Map.Entry;
  30 import java.util.Set;
  31 import java.util.concurrent.Future;
  32 import java.util.concurrent.TimeUnit;
  33 import org.apache.commons.lang3.SerializationUtils;
  34 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  35 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  36 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
  37 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  38 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
  39 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
  40 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
  41 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  42 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
  43 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
  44 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  45 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  46 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
  68 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  69 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  70 import org.opendaylight.yangtools.yang.common.RpcResult;
  71 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  73 import org.slf4j.Logger;
  74 import org.slf4j.LoggerFactory;
  75
  76 /**
  77  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
  78  *
  79  * @author Thomas Pantelis
  80  */
  81 public class ClusterAdminRpcService implements ClusterAdminService {
  82     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
  83
  84     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
  85
  86     private final DistributedDataStoreInterface configDataStore;
  87     private final DistributedDataStoreInterface operDataStore;
  88     private final BindingNormalizedNodeSerializer serializer;
  89
  90     public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
  91             DistributedDataStoreInterface operDataStore,
  92             BindingNormalizedNodeSerializer serializer) {
  93         this.configDataStore = configDataStore;
  94         this.operDataStore = operDataStore;
  95         this.serializer = serializer;
  96     }
  97
  98     @Override
  99     public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
 100         final String shardName = input.getShardName();
 101         if (Strings.isNullOrEmpty(shardName)) {
 102             return newFailedRpcResultFuture("A valid shard name must be specified");
 103         }
 104
 105         DataStoreType dataStoreType = input.getDataStoreType();
 106         if (dataStoreType == null) {
 107             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 108         }
 109
 110         LOG.info("Adding replica for shard {}", shardName);
 111
 112         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 113         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 114         Futures.addCallback(future, new FutureCallback<Success>() {
 115             @Override
 116             public void onSuccess(Success success) {
 117                 LOG.info("Successfully added replica for shard {}", shardName);
 118                 returnFuture.set(newSuccessfulResult());
 119             }
 120
 121             @Override
 122             public void onFailure(Throwable failure) {
 123                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 124                         returnFuture, failure);
 125             }
 126         });
 127
 128         return returnFuture;
 129     }
 130
 131     @Override
 132     public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
 133         final String shardName = input.getShardName();
 134         if (Strings.isNullOrEmpty(shardName)) {
 135             return newFailedRpcResultFuture("A valid shard name must be specified");
 136         }
 137
 138         DataStoreType dataStoreType = input.getDataStoreType();
 139         if (dataStoreType == null) {
 140             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 141         }
 142
 143         final String memberName = input.getMemberName();
 144         if (Strings.isNullOrEmpty(memberName)) {
 145             return newFailedRpcResultFuture("A valid member name must be specified");
 146         }
 147
 148         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 149
 150         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 151         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 152                 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
 153         Futures.addCallback(future, new FutureCallback<Success>() {
 154             @Override
 155             public void onSuccess(Success success) {
 156                 LOG.info("Successfully removed replica for shard {}", shardName);
 157                 returnFuture.set(newSuccessfulResult());
 158             }
 159
 160             @Override
 161             public void onFailure(Throwable failure) {
 162                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 163                         returnFuture, failure);
 164             }
 165         });
 166
 167         return returnFuture;
 168     }
 169
 170     @Override
 171     public Future<RpcResult<Void>> addPrefixShardReplica(final AddPrefixShardReplicaInput input) {
 172
 173         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 174         if (identifier == null) {
 175             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 176         }
 177
 178         final DataStoreType dataStoreType = input.getDataStoreType();
 179         if (dataStoreType == null) {
 180             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 181         }
 182
 183         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
 184
 185         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 186         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 187         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
 188         Futures.addCallback(future, new FutureCallback<Success>() {
 189             @Override
 190             public void onSuccess(Success success) {
 191                 LOG.info("Successfully added replica for shard {}", prefix);
 192                 returnFuture.set(newSuccessfulResult());
 193             }
 194
 195             @Override
 196             public void onFailure(Throwable failure) {
 197                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
 198                         returnFuture, failure);
 199             }
 200         });
 201
 202         return returnFuture;
 203     }
 204
 205     @Override
 206     public Future<RpcResult<Void>> removePrefixShardReplica(final RemovePrefixShardReplicaInput input) {
 207
 208         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 209         if (identifier == null) {
 210             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 211         }
 212
 213         final DataStoreType dataStoreType = input.getDataStoreType();
 214         if (dataStoreType == null) {
 215             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 216         }
 217
 218         final String memberName = input.getMemberName();
 219         if (Strings.isNullOrEmpty(memberName)) {
 220             return newFailedRpcResultFuture("A valid member name must be specified");
 221         }
 222
 223         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
 224                 identifier, memberName, dataStoreType);
 225         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 226
 227         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 228         final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 229                 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
 230         Futures.addCallback(future, new FutureCallback<Success>() {
 231             @Override
 232             public void onSuccess(final Success success) {
 233                 LOG.info("Successfully removed replica for shard {}", prefix);
 234                 returnFuture.set(newSuccessfulResult());
 235             }
 236
 237             @Override
 238             public void onFailure(final Throwable failure) {
 239                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
 240                         returnFuture, failure);
 241             }
 242         });
 243
 244         return returnFuture;
 245     }
 246
 247     @Override
 248     public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
 249         LOG.info("Adding replicas for all shards");
 250
 251         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 252         Function<String, Object> messageSupplier = shardName -> new AddShardReplica(shardName);
 253
 254         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 255         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 256
 257         return waitForShardResults(shardResultData, shardResults ->
 258                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 259                 "Failed to add replica");
 260     }
 261
 262
 263     @Override
 264     public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
 265         LOG.info("Removing replicas for all shards");
 266
 267         final String memberName = input.getMemberName();
 268         if (Strings.isNullOrEmpty(memberName)) {
 269             return newFailedRpcResultFuture("A valid member name must be specified");
 270         }
 271
 272         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 273         Function<String, Object> messageSupplier = shardName ->
 274                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 275
 276         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 277         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 278
 279         return waitForShardResults(shardResultData, shardResults ->
 280                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 281         "       Failed to remove replica");
 282     }
 283
 284     @Override
 285     public Future<RpcResult<Void>> changeMemberVotingStatesForShard(ChangeMemberVotingStatesForShardInput input) {
 286         final String shardName = input.getShardName();
 287         if (Strings.isNullOrEmpty(shardName)) {
 288             return newFailedRpcResultFuture("A valid shard name must be specified");
 289         }
 290
 291         DataStoreType dataStoreType = input.getDataStoreType();
 292         if (dataStoreType == null) {
 293             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 294         }
 295
 296         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 297         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 298             return newFailedRpcResultFuture("No member voting state input was specified");
 299         }
 300
 301         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 302                 memberVotingStates);
 303
 304         LOG.info("Change member voting states for shard {}: {}", shardName,
 305                 changeVotingStatus.getMeberVotingStatusMap());
 306
 307         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 308         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 309         Futures.addCallback(future, new FutureCallback<Success>() {
 310             @Override
 311             public void onSuccess(Success success) {
 312                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 313                 returnFuture.set(newSuccessfulResult());
 314             }
 315
 316             @Override
 317             public void onFailure(Throwable failure) {
 318                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 319                         returnFuture, failure);
 320             }
 321         });
 322
 323         return returnFuture;
 324     }
 325
 326     @Override
 327     public Future<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
 328             final ChangeMemberVotingStatesForAllShardsInput input) {
 329         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 330         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 331             return newFailedRpcResultFuture("No member voting state input was specified");
 332         }
 333
 334         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 335         Function<String, Object> messageSupplier = shardName ->
 336                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 337
 338         LOG.info("Change member voting states for all shards");
 339
 340         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 341         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 342
 343         return waitForShardResults(shardResultData, shardResults ->
 344                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 345                 "Failed to change member voting states");
 346     }
 347
 348     @Override
 349     public Future<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards() {
 350         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 351         Function<String, Object> messageSupplier = shardName ->
 352                 new FlipShardMembersVotingStatus(shardName);
 353
 354         LOG.info("Flip member voting states for all shards");
 355
 356         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 357         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 358
 359         return waitForShardResults(shardResultData, shardResults ->
 360                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 361                 "Failed to change member voting states");
 362     }
 363
 364     @Override
 365     public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
 366         LOG.debug("backupDatastore: {}", input);
 367
 368         if (Strings.isNullOrEmpty(input.getFilePath())) {
 369             return newFailedRpcResultFuture("A valid file path must be specified");
 370         }
 371
 372         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 373         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 374         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 375             @Override
 376             public void onSuccess(List<DatastoreSnapshot> snapshots) {
 377                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 378             }
 379
 380             @Override
 381             public void onFailure(Throwable failure) {
 382                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 383             }
 384         });
 385
 386         return returnFuture;
 387     }
 388
 389     private ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
 390             List<MemberVotingState> memberVotingStatus) {
 391         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
 392         for (MemberVotingState memberStatus: memberVotingStatus) {
 393             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
 394         }
 395
 396         ChangeShardMembersVotingStatus changeVotingStatus = new ChangeShardMembersVotingStatus(shardName,
 397                 serverVotingStatusMap);
 398         return changeVotingStatus;
 399     }
 400
 401     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 402             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 403             final Function<List<ShardResult>, T> resultDataSupplier,
 404             final String failureLogMsgPrefix) {
 405         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 406         final List<ShardResult> shardResults = new ArrayList<>();
 407         for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
 408             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 409                 @Override
 410                 public void onSuccess(Success result) {
 411                     synchronized (shardResults) {
 412                         ShardResultBuilder shardResult = entry.getValue();
 413                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 414                                 shardResult.getDataStoreType());
 415                         shardResults.add(shardResult.setSucceeded(true).build());
 416                         checkIfComplete();
 417                     }
 418                 }
 419
 420                 @Override
 421                 public void onFailure(Throwable failure) {
 422                     synchronized (shardResults) {
 423                         ShardResultBuilder shardResult = entry.getValue();
 424                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 425                                 shardResult.getDataStoreType(), failure);
 426                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 427                                 Throwables.getRootCause(failure).getMessage()).build());
 428                         checkIfComplete();
 429                     }
 430                 }
 431
 432                 void checkIfComplete() {
 433                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 434                     if (shardResults.size() == shardResultData.size()) {
 435                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 436                     }
 437                 }
 438             });
 439         }
 440         return returnFuture;
 441     }
 442
 443     private <T> void sendMessageToManagerForConfiguredShards(DataStoreType dataStoreType,
 444             List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 445             Function<String, Object> messageSupplier) {
 446         ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
 447                 : operDataStore.getActorContext();
 448         Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
 449
 450         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
 451
 452         for (String shardName: allShardNames) {
 453             ListenableFuture<T> future = this.<T>ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
 454                     SHARD_MGR_TIMEOUT);
 455             shardResultData.add(new SimpleEntry<>(future,
 456                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 457         }
 458     }
 459
 460     @SuppressWarnings("unchecked")
 461     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
 462         Timeout timeout = SHARD_MGR_TIMEOUT;
 463         ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
 464         ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
 465
 466         return Futures.allAsList(configFuture, operFuture);
 467     }
 468
 469     private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
 470         ActorRef shardManager = dataStoreType == DataStoreType.Config
 471                 ? configDataStore.getActorContext().getShardManager()
 472                         : operDataStore.getActorContext().getShardManager();
 473         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 474     }
 475
 476     @SuppressWarnings("checkstyle:IllegalCatch")
 477     private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
 478             SettableFuture<RpcResult<Void>> returnFuture) {
 479         try (FileOutputStream fos = new FileOutputStream(fileName)) {
 480             SerializationUtils.serialize(snapshots, fos);
 481
 482             returnFuture.set(newSuccessfulResult());
 483             LOG.info("Successfully backed up datastore to file {}", fileName);
 484         } catch (IOException | RuntimeException e) {
 485             onDatastoreBackupFailure(fileName, returnFuture, e);
 486         }
 487     }
 488
 489     private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
 490             Throwable failure) {
 491         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 492     }
 493
 494     private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
 495             Throwable failure) {
 496         LOG.error(msg, failure);
 497         returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 498                 failure.getMessage())).build());
 499     }
 500
 501     private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
 502         final SettableFuture<T> returnFuture = SettableFuture.create();
 503
 504         @SuppressWarnings("unchecked")
 505         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 506         askFuture.onComplete(new OnComplete<T>() {
 507             @Override
 508             public void onComplete(Throwable failure, T resp) {
 509                 if (failure != null) {
 510                     returnFuture.setException(failure);
 511                 } else {
 512                     returnFuture.set(resp);
 513                 }
 514             }
 515         }, configDataStore.getActorContext().getClientDispatcher());
 516
 517         return returnFuture;
 518     }
 519
 520     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
 521         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 522     }
 523
 524     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
 525         return newFailedRpcResultBuilder(message, null);
 526     }
 527
 528     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
 529         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 530     }
 531
 532     private static RpcResult<Void> newSuccessfulResult() {
 533         return newSuccessfulResult((Void)null);
 534     }
 535
 536     private static <T> RpcResult<T> newSuccessfulResult(T data) {
 537         return RpcResultBuilder.<T>success(data).build();
 538     }
 539 }