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