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.Strings;
  16 import com.google.common.base.Throwables;
  17 import com.google.common.util.concurrent.FutureCallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import com.google.common.util.concurrent.MoreExecutors;
  21 import com.google.common.util.concurrent.SettableFuture;
  22 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  23 import java.io.FileOutputStream;
  24 import java.io.IOException;
  25 import java.util.AbstractMap.SimpleEntry;
  26 import java.util.ArrayList;
  27 import java.util.HashMap;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.Map.Entry;
  31 import java.util.Set;
  32 import java.util.concurrent.TimeUnit;
  33 import java.util.function.Function;
  34 import org.apache.commons.lang3.SerializationUtils;
  35 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  36 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  37 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
  38 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  39 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
  40 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
  41 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
  42 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
  43 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
  44 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
  45 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  46 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
  47 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
  48 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
  49 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  50 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  51 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutput;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutputBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
  62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
  63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
  64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
  65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
  66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
  67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
  68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
  69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
  70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutput;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutputBuilder;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
  96 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  97 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  98 import org.opendaylight.yangtools.yang.common.RpcResult;
  99 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
 100 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
 101 import org.slf4j.Logger;
 102 import org.slf4j.LoggerFactory;
 103
 104 /**
 105  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
 106  *
 107  * @author Thomas Pantelis
 108  */
 109 public class ClusterAdminRpcService implements ClusterAdminService {
 110     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
 111
 112     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
 113
 114     private final DistributedDataStoreInterface configDataStore;
 115     private final DistributedDataStoreInterface operDataStore;
 116     private final BindingNormalizedNodeSerializer serializer;
 117     private final Timeout makeLeaderLocalTimeout;
 118
 119     public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
 120             final DistributedDataStoreInterface operDataStore,
 121             final BindingNormalizedNodeSerializer serializer) {
 122         this.configDataStore = configDataStore;
 123         this.operDataStore = operDataStore;
 124         this.serializer = serializer;
 125
 126         this.makeLeaderLocalTimeout =
 127                 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
 128                         .getShardLeaderElectionTimeout().duration().$times(2));
 129     }
 130
 131     @Override
 132     public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput 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         LOG.info("Adding replica for shard {}", shardName);
 144
 145         final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
 146         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 147         Futures.addCallback(future, new FutureCallback<Success>() {
 148             @Override
 149             public void onSuccess(final Success success) {
 150                 LOG.info("Successfully added replica for shard {}", shardName);
 151                 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
 152             }
 153
 154             @Override
 155             public void onFailure(final Throwable failure) {
 156                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 157                         returnFuture, failure);
 158             }
 159         }, MoreExecutors.directExecutor());
 160
 161         return returnFuture;
 162     }
 163
 164     @Override
 165     public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
 166             final RemoveShardReplicaInput input) {
 167         final String shardName = input.getShardName();
 168         if (Strings.isNullOrEmpty(shardName)) {
 169             return newFailedRpcResultFuture("A valid shard name must be specified");
 170         }
 171
 172         DataStoreType dataStoreType = input.getDataStoreType();
 173         if (dataStoreType == null) {
 174             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 175         }
 176
 177         final String memberName = input.getMemberName();
 178         if (Strings.isNullOrEmpty(memberName)) {
 179             return newFailedRpcResultFuture("A valid member name must be specified");
 180         }
 181
 182         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 183
 184         final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
 185         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 186                 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
 187         Futures.addCallback(future, new FutureCallback<Success>() {
 188             @Override
 189             public void onSuccess(final Success success) {
 190                 LOG.info("Successfully removed replica for shard {}", shardName);
 191                 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
 192             }
 193
 194             @Override
 195             public void onFailure(final Throwable failure) {
 196                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 197                         returnFuture, failure);
 198             }
 199         }, MoreExecutors.directExecutor());
 200
 201         return returnFuture;
 202     }
 203
 204     @Override
 205     public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
 206         final String shardName = input.getShardName();
 207         if (Strings.isNullOrEmpty(shardName)) {
 208             return newFailedRpcResultFuture("A valid shard name must be specified");
 209         }
 210
 211         DataStoreType dataStoreType = input.getDataStoreType();
 212         if (dataStoreType == null) {
 213             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 214         }
 215
 216         ActorUtils actorUtils = dataStoreType == DataStoreType.Config
 217                 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
 218
 219         LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
 220                 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
 221
 222         final scala.concurrent.Future<ActorRef> localShardReply =
 223                 actorUtils.findLocalShardAsync(shardName);
 224
 225         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
 226         localShardReply.onComplete(new OnComplete<ActorRef>() {
 227             @Override
 228             public void onComplete(final Throwable failure, final ActorRef actorRef) {
 229                 if (failure != null) {
 230                     LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
 231                             + " local shard.", shardName, dataStoreType, failure);
 232                     makeLeaderLocalAsk.failure(failure);
 233                 } else {
 234                     makeLeaderLocalAsk
 235                             .completeWith(actorUtils
 236                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
 237                 }
 238             }
 239         }, actorUtils.getClientDispatcher());
 240
 241         final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
 242         makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
 243             @Override
 244             public void onComplete(final Throwable failure, final Object success) {
 245                 if (failure != null) {
 246                     LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
 247                     future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
 248                             "leadership transfer failed", failure).build());
 249                     return;
 250                 }
 251
 252                 LOG.debug("Leadership transfer complete");
 253                 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
 254             }
 255         }, actorUtils.getClientDispatcher());
 256
 257         return future;
 258     }
 259
 260     @Override
 261     public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
 262             final AddPrefixShardReplicaInput input) {
 263
 264         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 265         if (identifier == null) {
 266             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 267         }
 268
 269         final DataStoreType dataStoreType = input.getDataStoreType();
 270         if (dataStoreType == null) {
 271             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 272         }
 273
 274         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
 275
 276         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 277         final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 278         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
 279         Futures.addCallback(future, new FutureCallback<Success>() {
 280             @Override
 281             public void onSuccess(final Success success) {
 282                 LOG.info("Successfully added replica for shard {}", prefix);
 283                 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
 284             }
 285
 286             @Override
 287             public void onFailure(final Throwable failure) {
 288                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
 289                         returnFuture, failure);
 290             }
 291         }, MoreExecutors.directExecutor());
 292
 293         return returnFuture;
 294     }
 295
 296     @Override
 297     public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
 298             final RemovePrefixShardReplicaInput input) {
 299
 300         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 301         if (identifier == null) {
 302             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 303         }
 304
 305         final DataStoreType dataStoreType = input.getDataStoreType();
 306         if (dataStoreType == null) {
 307             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 308         }
 309
 310         final String memberName = input.getMemberName();
 311         if (Strings.isNullOrEmpty(memberName)) {
 312             return newFailedRpcResultFuture("A valid member name must be specified");
 313         }
 314
 315         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
 316                 identifier, memberName, dataStoreType);
 317         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 318
 319         final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 320         final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 321                 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
 322         Futures.addCallback(future, new FutureCallback<Success>() {
 323             @Override
 324             public void onSuccess(final Success success) {
 325                 LOG.info("Successfully removed replica for shard {}", prefix);
 326                 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
 327             }
 328
 329             @Override
 330             public void onFailure(final Throwable failure) {
 331                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
 332                         returnFuture, failure);
 333             }
 334         }, MoreExecutors.directExecutor());
 335
 336         return returnFuture;
 337     }
 338
 339     @Override
 340     public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
 341             final AddReplicasForAllShardsInput input) {
 342         LOG.info("Adding replicas for all shards");
 343
 344         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 345
 346         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
 347         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
 348
 349         return waitForShardResults(shardResultData, shardResults ->
 350                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 351                 "Failed to add replica");
 352     }
 353
 354
 355     @Override
 356     public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
 357             final RemoveAllShardReplicasInput input) {
 358         LOG.info("Removing replicas for all shards");
 359
 360         final String memberName = input.getMemberName();
 361         if (Strings.isNullOrEmpty(memberName)) {
 362             return newFailedRpcResultFuture("A valid member name must be specified");
 363         }
 364
 365         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 366         Function<String, Object> messageSupplier = shardName ->
 367                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 368
 369         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 370         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 371
 372         return waitForShardResults(shardResultData, shardResults ->
 373                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 374         "       Failed to remove replica");
 375     }
 376
 377     @Override
 378     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
 379             final ChangeMemberVotingStatesForShardInput input) {
 380         final String shardName = input.getShardName();
 381         if (Strings.isNullOrEmpty(shardName)) {
 382             return newFailedRpcResultFuture("A valid shard name must be specified");
 383         }
 384
 385         DataStoreType dataStoreType = input.getDataStoreType();
 386         if (dataStoreType == null) {
 387             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 388         }
 389
 390         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 391         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 392             return newFailedRpcResultFuture("No member voting state input was specified");
 393         }
 394
 395         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 396                 memberVotingStates);
 397
 398         LOG.info("Change member voting states for shard {}: {}", shardName,
 399                 changeVotingStatus.getMeberVotingStatusMap());
 400
 401         final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
 402         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 403         Futures.addCallback(future, new FutureCallback<Success>() {
 404             @Override
 405             public void onSuccess(final Success success) {
 406                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 407                 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
 408             }
 409
 410             @Override
 411             public void onFailure(final Throwable failure) {
 412                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 413                         returnFuture, failure);
 414             }
 415         }, MoreExecutors.directExecutor());
 416
 417         return returnFuture;
 418     }
 419
 420     @Override
 421     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
 422             final ChangeMemberVotingStatesForAllShardsInput input) {
 423         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 424         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 425             return newFailedRpcResultFuture("No member voting state input was specified");
 426         }
 427
 428         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 429         Function<String, Object> messageSupplier = shardName ->
 430                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 431
 432         LOG.info("Change member voting states for all shards");
 433
 434         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 435         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 436
 437         return waitForShardResults(shardResultData, shardResults ->
 438                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 439                 "Failed to change member voting states");
 440     }
 441
 442     @Override
 443     public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
 444             final FlipMemberVotingStatesForAllShardsInput input) {
 445         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 446         Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
 447
 448         LOG.info("Flip member voting states for all shards");
 449
 450         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 451         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 452
 453         return waitForShardResults(shardResultData, shardResults ->
 454                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 455                 "Failed to change member voting states");
 456     }
 457
 458     @Override
 459     public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
 460         final String shardName = input.getShardName();
 461         if (Strings.isNullOrEmpty(shardName)) {
 462             return newFailedRpcResultFuture("A valid shard name must be specified");
 463         }
 464
 465         DataStoreType dataStoreType = input.getDataStoreType();
 466         if (dataStoreType == null) {
 467             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 468         }
 469
 470         LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
 471
 472         final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
 473         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 474                 new GetShardRole(shardName));
 475         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 476             @Override
 477             public void onSuccess(final GetShardRoleReply reply) {
 478                 if (reply == null) {
 479                     returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 480                             "No Shard role present. Please retry..").build());
 481                     return;
 482                 }
 483                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 484                 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
 485                 if (reply.getRole() != null) {
 486                     builder.setRole(reply.getRole());
 487                 }
 488                 returnFuture.set(newSuccessfulResult(builder.build()));
 489             }
 490
 491             @Override
 492             public void onFailure(final Throwable failure) {
 493                 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 494                         "Failed to get shard role.", failure).build());
 495             }
 496         }, MoreExecutors.directExecutor());
 497
 498         return returnFuture;
 499     }
 500
 501     @Override
 502     public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
 503             final GetPrefixShardRoleInput input) {
 504         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 505         if (identifier == null) {
 506             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 507         }
 508
 509         final DataStoreType dataStoreType = input.getDataStoreType();
 510         if (dataStoreType == null) {
 511             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 512         }
 513
 514         LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
 515
 516         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 517         final String shardName = ClusterUtils.getCleanShardName(prefix);
 518         final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
 519         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 520                 new GetShardRole(shardName));
 521         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 522             @Override
 523             public void onSuccess(final GetShardRoleReply reply) {
 524                 if (reply == null) {
 525                     returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 526                             "No Shard role present. Please retry..").build());
 527                     return;
 528                 }
 529
 530                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 531                 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
 532                 if (reply.getRole() != null) {
 533                     builder.setRole(reply.getRole());
 534                 }
 535                 returnFuture.set(newSuccessfulResult(builder.build()));
 536             }
 537
 538             @Override
 539             public void onFailure(final Throwable failure) {
 540                 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 541                         "Failed to get shard role.", failure).build());
 542             }
 543         }, MoreExecutors.directExecutor());
 544
 545         return returnFuture;
 546     }
 547
 548     @Override
 549     public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
 550         LOG.debug("backupDatastore: {}", input);
 551
 552         if (Strings.isNullOrEmpty(input.getFilePath())) {
 553             return newFailedRpcResultFuture("A valid file path must be specified");
 554         }
 555
 556         final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
 557         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 558         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 559             @Override
 560             public void onSuccess(final List<DatastoreSnapshot> snapshots) {
 561                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 562             }
 563
 564             @Override
 565             public void onFailure(final Throwable failure) {
 566                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 567             }
 568         }, MoreExecutors.directExecutor());
 569
 570         return returnFuture;
 571     }
 572
 573     private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
 574             final List<MemberVotingState> memberVotingStatus) {
 575         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
 576         for (MemberVotingState memberStatus: memberVotingStatus) {
 577             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
 578         }
 579         return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
 580     }
 581
 582     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 583             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 584             final Function<List<ShardResult>, T> resultDataSupplier,
 585             final String failureLogMsgPrefix) {
 586         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 587         final List<ShardResult> shardResults = new ArrayList<>();
 588         for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
 589             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 590                 @Override
 591                 public void onSuccess(final Success result) {
 592                     synchronized (shardResults) {
 593                         ShardResultBuilder shardResult = entry.getValue();
 594                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 595                                 shardResult.getDataStoreType());
 596                         shardResults.add(shardResult.setSucceeded(true).build());
 597                         checkIfComplete();
 598                     }
 599                 }
 600
 601                 @Override
 602                 public void onFailure(final Throwable failure) {
 603                     synchronized (shardResults) {
 604                         ShardResultBuilder shardResult = entry.getValue();
 605                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 606                                 shardResult.getDataStoreType(), failure);
 607                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 608                                 Throwables.getRootCause(failure).getMessage()).build());
 609                         checkIfComplete();
 610                     }
 611                 }
 612
 613                 void checkIfComplete() {
 614                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 615                     if (shardResults.size() == shardResultData.size()) {
 616                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 617                     }
 618                 }
 619             }, MoreExecutors.directExecutor());
 620         }
 621         return returnFuture;
 622     }
 623
 624     private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
 625             final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 626             final Function<String, Object> messageSupplier) {
 627         ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
 628                 : operDataStore.getActorUtils();
 629         Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
 630
 631         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
 632
 633         for (String shardName: allShardNames) {
 634             ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
 635                                                   SHARD_MGR_TIMEOUT);
 636             shardResultData.add(new SimpleEntry<>(future,
 637                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 638         }
 639     }
 640
 641     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
 642         Timeout timeout = SHARD_MGR_TIMEOUT;
 643         ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
 644         ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
 645
 646         return Futures.allAsList(configFuture, operFuture);
 647     }
 648
 649     private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
 650         ActorRef shardManager = dataStoreType == DataStoreType.Config
 651                 ? configDataStore.getActorUtils().getShardManager()
 652                         : operDataStore.getActorUtils().getShardManager();
 653         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 654     }
 655
 656     @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
 657             justification = "https://github.com/spotbugs/spotbugs/issues/811")
 658     @SuppressWarnings("checkstyle:IllegalCatch")
 659     private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
 660             final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
 661         try (FileOutputStream fos = new FileOutputStream(fileName)) {
 662             SerializationUtils.serialize(snapshots, fos);
 663
 664             returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
 665             LOG.info("Successfully backed up datastore to file {}", fileName);
 666         } catch (IOException | RuntimeException e) {
 667             onDatastoreBackupFailure(fileName, returnFuture, e);
 668         }
 669     }
 670
 671     private static <T> void onDatastoreBackupFailure(final String fileName,
 672             final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
 673         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 674     }
 675
 676     @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
 677     private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
 678             final Throwable failure) {
 679         LOG.error("{}", msg, failure);
 680         returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 681                 failure.getMessage())).build());
 682     }
 683
 684     private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
 685         final SettableFuture<T> returnFuture = SettableFuture.create();
 686
 687         @SuppressWarnings("unchecked")
 688         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 689         askFuture.onComplete(new OnComplete<T>() {
 690             @Override
 691             public void onComplete(final Throwable failure, final T resp) {
 692                 if (failure != null) {
 693                     returnFuture.setException(failure);
 694                 } else {
 695                     returnFuture.set(resp);
 696                 }
 697             }
 698         }, configDataStore.getActorUtils().getClientDispatcher());
 699
 700         return returnFuture;
 701     }
 702
 703     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
 704         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 705     }
 706
 707     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
 708         return newFailedRpcResultBuilder(message, null);
 709     }
 710
 711     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
 712         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 713     }
 714
 715     private static <T> RpcResult<T> newSuccessfulResult(final T data) {
 716         return RpcResultBuilder.success(data).build();
 717     }
 718 }