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
  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;
  30 import java.util.Map;
  31 import java.util.Map.Entry;
  32 import java.util.Set;
  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;
  88
  89 /**
  90  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
  91  *
  92  * @author Thomas Pantelis
  93  */
  94 public class ClusterAdminRpcService implements ClusterAdminService {
  95     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
  96
  97     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
  98
  99     private final DistributedDataStoreInterface configDataStore;
 100     private final DistributedDataStoreInterface operDataStore;
 101     private final BindingNormalizedNodeSerializer serializer;
 102     private final Timeout makeLeaderLocalTimeout;
 103
 104     public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
 105             DistributedDataStoreInterface operDataStore,
 106             BindingNormalizedNodeSerializer serializer) {
 107         this.configDataStore = configDataStore;
 108         this.operDataStore = operDataStore;
 109         this.serializer = serializer;
 110
 111         this.makeLeaderLocalTimeout =
 112                 new Timeout(configDataStore.getActorContext().getDatastoreContext()
 113                         .getShardLeaderElectionTimeout().duration().$times(2));
 114     }
 115
 116     @Override
 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");
 121         }
 122
 123         DataStoreType dataStoreType = input.getDataStoreType();
 124         if (dataStoreType == null) {
 125             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 126         }
 127
 128         LOG.info("Adding replica for shard {}", shardName);
 129
 130         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 131         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 132         Futures.addCallback(future, new FutureCallback<Success>() {
 133             @Override
 134             public void onSuccess(Success success) {
 135                 LOG.info("Successfully added replica for shard {}", shardName);
 136                 returnFuture.set(newSuccessfulResult());
 137             }
 138
 139             @Override
 140             public void onFailure(Throwable failure) {
 141                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 142                         returnFuture, failure);
 143             }
 144         }, MoreExecutors.directExecutor());
 145
 146         return returnFuture;
 147     }
 148
 149     @Override
 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");
 154         }
 155
 156         DataStoreType dataStoreType = input.getDataStoreType();
 157         if (dataStoreType == null) {
 158             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 159         }
 160
 161         final String memberName = input.getMemberName();
 162         if (Strings.isNullOrEmpty(memberName)) {
 163             return newFailedRpcResultFuture("A valid member name must be specified");
 164         }
 165
 166         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 167
 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>() {
 172             @Override
 173             public void onSuccess(Success success) {
 174                 LOG.info("Successfully removed replica for shard {}", shardName);
 175                 returnFuture.set(newSuccessfulResult());
 176             }
 177
 178             @Override
 179             public void onFailure(Throwable failure) {
 180                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 181                         returnFuture, failure);
 182             }
 183         }, MoreExecutors.directExecutor());
 184
 185         return returnFuture;
 186     }
 187
 188     @Override
 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");
 193         }
 194
 195         DataStoreType dataStoreType = input.getDataStoreType();
 196         if (dataStoreType == null) {
 197             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 198         }
 199
 200         ActorContext actorContext = dataStoreType == DataStoreType.Config
 201                 ? configDataStore.getActorContext()
 202                 : operDataStore.getActorContext();
 203
 204         LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
 205                 actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
 206
 207         final scala.concurrent.Future<ActorRef> localShardReply =
 208                 actorContext.findLocalShardAsync(shardName);
 209
 210         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
 211         localShardReply.onComplete(new OnComplete<ActorRef>() {
 212             @Override
 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);
 218                 } else {
 219                     makeLeaderLocalAsk
 220                             .completeWith(actorContext
 221                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
 222                 }
 223             }
 224         }, actorContext.getClientDispatcher());
 225
 226         final SettableFuture<RpcResult<Void>> future = SettableFuture.create();
 227         makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
 228             @Override
 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());
 234                     return;
 235                 }
 236
 237                 LOG.debug("Leadership transfer complete");
 238                 future.set(RpcResultBuilder.<Void>success().build());
 239             }
 240         }, actorContext.getClientDispatcher());
 241
 242         return future;
 243     }
 244
 245     @Override
 246     public Future<RpcResult<Void>> addPrefixShardReplica(final AddPrefixShardReplicaInput input) {
 247
 248         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 249         if (identifier == null) {
 250             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 251         }
 252
 253         final DataStoreType dataStoreType = input.getDataStoreType();
 254         if (dataStoreType == null) {
 255             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 256         }
 257
 258         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
 259
 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>() {
 264             @Override
 265             public void onSuccess(Success success) {
 266                 LOG.info("Successfully added replica for shard {}", prefix);
 267                 returnFuture.set(newSuccessfulResult());
 268             }
 269
 270             @Override
 271             public void onFailure(Throwable failure) {
 272                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
 273                         returnFuture, failure);
 274             }
 275         }, MoreExecutors.directExecutor());
 276
 277         return returnFuture;
 278     }
 279
 280     @Override
 281     public Future<RpcResult<Void>> removePrefixShardReplica(final RemovePrefixShardReplicaInput input) {
 282
 283         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 284         if (identifier == null) {
 285             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 286         }
 287
 288         final DataStoreType dataStoreType = input.getDataStoreType();
 289         if (dataStoreType == null) {
 290             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 291         }
 292
 293         final String memberName = input.getMemberName();
 294         if (Strings.isNullOrEmpty(memberName)) {
 295             return newFailedRpcResultFuture("A valid member name must be specified");
 296         }
 297
 298         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
 299                 identifier, memberName, dataStoreType);
 300         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 301
 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>() {
 306             @Override
 307             public void onSuccess(final Success success) {
 308                 LOG.info("Successfully removed replica for shard {}", prefix);
 309                 returnFuture.set(newSuccessfulResult());
 310             }
 311
 312             @Override
 313             public void onFailure(final Throwable failure) {
 314                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
 315                         returnFuture, failure);
 316             }
 317         }, MoreExecutors.directExecutor());
 318
 319         return returnFuture;
 320     }
 321
 322     @Override
 323     public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
 324         LOG.info("Adding replicas for all shards");
 325
 326         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 327         Function<String, Object> messageSupplier = AddShardReplica::new;
 328
 329         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 330         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 331
 332         return waitForShardResults(shardResultData, shardResults ->
 333                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 334                 "Failed to add replica");
 335     }
 336
 337
 338     @Override
 339     public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
 340         LOG.info("Removing replicas for all shards");
 341
 342         final String memberName = input.getMemberName();
 343         if (Strings.isNullOrEmpty(memberName)) {
 344             return newFailedRpcResultFuture("A valid member name must be specified");
 345         }
 346
 347         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 348         Function<String, Object> messageSupplier = shardName ->
 349                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 350
 351         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 352         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 353
 354         return waitForShardResults(shardResultData, shardResults ->
 355                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 356         "       Failed to remove replica");
 357     }
 358
 359     @Override
 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");
 364         }
 365
 366         DataStoreType dataStoreType = input.getDataStoreType();
 367         if (dataStoreType == null) {
 368             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 369         }
 370
 371         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 372         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 373             return newFailedRpcResultFuture("No member voting state input was specified");
 374         }
 375
 376         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 377                 memberVotingStates);
 378
 379         LOG.info("Change member voting states for shard {}: {}", shardName,
 380                 changeVotingStatus.getMeberVotingStatusMap());
 381
 382         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 383         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 384         Futures.addCallback(future, new FutureCallback<Success>() {
 385             @Override
 386             public void onSuccess(Success success) {
 387                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 388                 returnFuture.set(newSuccessfulResult());
 389             }
 390
 391             @Override
 392             public void onFailure(Throwable failure) {
 393                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 394                         returnFuture, failure);
 395             }
 396         }, MoreExecutors.directExecutor());
 397
 398         return returnFuture;
 399     }
 400
 401     @Override
 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");
 407         }
 408
 409         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 410         Function<String, Object> messageSupplier = shardName ->
 411                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 412
 413         LOG.info("Change member voting states for all shards");
 414
 415         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 416         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 417
 418         return waitForShardResults(shardResultData, shardResults ->
 419                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 420                 "Failed to change member voting states");
 421     }
 422
 423     @Override
 424     public Future<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards() {
 425         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 426         Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
 427
 428         LOG.info("Flip member voting states for all shards");
 429
 430         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 431         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 432
 433         return waitForShardResults(shardResultData, shardResults ->
 434                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 435                 "Failed to change member voting states");
 436     }
 437
 438     @Override
 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");
 443         }
 444
 445         DataStoreType dataStoreType = input.getDataStoreType();
 446         if (dataStoreType == null) {
 447             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 448         }
 449
 450         LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
 451
 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>() {
 456             @Override
 457             public void onSuccess(final GetShardRoleReply reply) {
 458                 if (reply == null) {
 459                     returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 460                             "No Shard role present. Please retry..").build());
 461                     return;
 462                 }
 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());
 467                 }
 468                 returnFuture.set(newSuccessfulResult(builder.build()));
 469             }
 470
 471             @Override
 472             public void onFailure(final Throwable failure) {
 473                 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 474                         "Failed to get shard role.", failure).build());
 475             }
 476         }, MoreExecutors.directExecutor());
 477
 478         return returnFuture;
 479     }
 480
 481     @Override
 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");
 486         }
 487
 488         final DataStoreType dataStoreType = input.getDataStoreType();
 489         if (dataStoreType == null) {
 490             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 491         }
 492
 493         LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
 494
 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>() {
 501             @Override
 502             public void onSuccess(final GetShardRoleReply reply) {
 503                 if (reply == null) {
 504                     returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 505                             "No Shard role present. Please retry..").build());
 506                     return;
 507                 }
 508
 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());
 513                 }
 514                 returnFuture.set(newSuccessfulResult(builder.build()));
 515             }
 516
 517             @Override
 518             public void onFailure(final Throwable failure) {
 519                 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 520                         "Failed to get shard role.", failure).build());
 521             }
 522         }, MoreExecutors.directExecutor());
 523
 524         return returnFuture;
 525     }
 526
 527     @Override
 528     public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
 529         LOG.debug("backupDatastore: {}", input);
 530
 531         if (Strings.isNullOrEmpty(input.getFilePath())) {
 532             return newFailedRpcResultFuture("A valid file path must be specified");
 533         }
 534
 535         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 536         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 537         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 538             @Override
 539             public void onSuccess(List<DatastoreSnapshot> snapshots) {
 540                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 541             }
 542
 543             @Override
 544             public void onFailure(Throwable failure) {
 545                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 546             }
 547         }, MoreExecutors.directExecutor());
 548
 549         return returnFuture;
 550     }
 551
 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());
 557         }
 558         return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
 559     }
 560
 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>() {
 569                 @Override
 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());
 576                         checkIfComplete();
 577                     }
 578                 }
 579
 580                 @Override
 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());
 588                         checkIfComplete();
 589                     }
 590                 }
 591
 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)));
 596                     }
 597                 }
 598             }, MoreExecutors.directExecutor());
 599         }
 600         return returnFuture;
 601     }
 602
 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();
 609
 610         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
 611
 612         for (String shardName: allShardNames) {
 613             ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
 614                                                   SHARD_MGR_TIMEOUT);
 615             shardResultData.add(new SimpleEntry<>(future,
 616                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 617         }
 618     }
 619
 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);
 625
 626         return Futures.allAsList(configFuture, operFuture);
 627     }
 628
 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);
 634     }
 635
 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);
 641
 642             returnFuture.set(newSuccessfulResult());
 643             LOG.info("Successfully backed up datastore to file {}", fileName);
 644         } catch (IOException | RuntimeException e) {
 645             onDatastoreBackupFailure(fileName, returnFuture, e);
 646         }
 647     }
 648
 649     private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
 650             Throwable failure) {
 651         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 652     }
 653
 654     private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
 655             Throwable failure) {
 656         LOG.error(msg, failure);
 657         returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 658                 failure.getMessage())).build());
 659     }
 660
 661     private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
 662         final SettableFuture<T> returnFuture = SettableFuture.create();
 663
 664         @SuppressWarnings("unchecked")
 665         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 666         askFuture.onComplete(new OnComplete<T>() {
 667             @Override
 668             public void onComplete(Throwable failure, T resp) {
 669                 if (failure != null) {
 670                     returnFuture.setException(failure);
 671                 } else {
 672                     returnFuture.set(resp);
 673                 }
 674             }
 675         }, configDataStore.getActorContext().getClientDispatcher());
 676
 677         return returnFuture;
 678     }
 679
 680     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
 681         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 682     }
 683
 684     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
 685         return newFailedRpcResultBuilder(message, null);
 686     }
 687
 688     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
 689         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 690     }
 691
 692     private static RpcResult<Void> newSuccessfulResult() {
 693         return newSuccessfulResult(null);
 694     }
 695
 696     private static <T> RpcResult<T> newSuccessfulResult(T data) {
 697         return RpcResultBuilder.success(data).build();
 698     }
 699 }