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 edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  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.TimeUnit;
  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.ActorContext;
  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.getActorContext().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         ActorContext actorContext = dataStoreType == DataStoreType.Config
 217                 ? configDataStore.getActorContext()
 218                 : operDataStore.getActorContext();
 219
 220         LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
 221                 actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
 222
 223         final scala.concurrent.Future<ActorRef> localShardReply =
 224                 actorContext.findLocalShardAsync(shardName);
 225
 226         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
 227         localShardReply.onComplete(new OnComplete<ActorRef>() {
 228             @Override
 229             public void onComplete(final Throwable failure, final ActorRef actorRef) {
 230                 if (failure != null) {
 231                     LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
 232                             + " local shard.", shardName, dataStoreType, failure);
 233                     makeLeaderLocalAsk.failure(failure);
 234                 } else {
 235                     makeLeaderLocalAsk
 236                             .completeWith(actorContext
 237                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
 238                 }
 239             }
 240         }, actorContext.getClientDispatcher());
 241
 242         final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
 243         makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
 244             @Override
 245             public void onComplete(final Throwable failure, final Object success) {
 246                 if (failure != null) {
 247                     LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
 248                     future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
 249                             "leadership transfer failed", failure).build());
 250                     return;
 251                 }
 252
 253                 LOG.debug("Leadership transfer complete");
 254                 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
 255             }
 256         }, actorContext.getClientDispatcher());
 257
 258         return future;
 259     }
 260
 261     @Override
 262     public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
 263             final AddPrefixShardReplicaInput input) {
 264
 265         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 266         if (identifier == null) {
 267             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 268         }
 269
 270         final DataStoreType dataStoreType = input.getDataStoreType();
 271         if (dataStoreType == null) {
 272             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 273         }
 274
 275         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
 276
 277         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 278         final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 279         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
 280         Futures.addCallback(future, new FutureCallback<Success>() {
 281             @Override
 282             public void onSuccess(final Success success) {
 283                 LOG.info("Successfully added replica for shard {}", prefix);
 284                 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
 285             }
 286
 287             @Override
 288             public void onFailure(final Throwable failure) {
 289                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
 290                         returnFuture, failure);
 291             }
 292         }, MoreExecutors.directExecutor());
 293
 294         return returnFuture;
 295     }
 296
 297     @Override
 298     public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
 299             final RemovePrefixShardReplicaInput input) {
 300
 301         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 302         if (identifier == null) {
 303             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 304         }
 305
 306         final DataStoreType dataStoreType = input.getDataStoreType();
 307         if (dataStoreType == null) {
 308             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 309         }
 310
 311         final String memberName = input.getMemberName();
 312         if (Strings.isNullOrEmpty(memberName)) {
 313             return newFailedRpcResultFuture("A valid member name must be specified");
 314         }
 315
 316         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
 317                 identifier, memberName, dataStoreType);
 318         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 319
 320         final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
 321         final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 322                 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
 323         Futures.addCallback(future, new FutureCallback<Success>() {
 324             @Override
 325             public void onSuccess(final Success success) {
 326                 LOG.info("Successfully removed replica for shard {}", prefix);
 327                 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
 328             }
 329
 330             @Override
 331             public void onFailure(final Throwable failure) {
 332                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
 333                         returnFuture, failure);
 334             }
 335         }, MoreExecutors.directExecutor());
 336
 337         return returnFuture;
 338     }
 339
 340     @Override
 341     public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
 342             final AddReplicasForAllShardsInput input) {
 343         LOG.info("Adding replicas for all shards");
 344
 345         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 346         Function<String, Object> messageSupplier = AddShardReplica::new;
 347
 348         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 349         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 350
 351         return waitForShardResults(shardResultData, shardResults ->
 352                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 353                 "Failed to add replica");
 354     }
 355
 356
 357     @Override
 358     public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
 359             final RemoveAllShardReplicasInput input) {
 360         LOG.info("Removing replicas for all shards");
 361
 362         final String memberName = input.getMemberName();
 363         if (Strings.isNullOrEmpty(memberName)) {
 364             return newFailedRpcResultFuture("A valid member name must be specified");
 365         }
 366
 367         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 368         Function<String, Object> messageSupplier = shardName ->
 369                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 370
 371         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 372         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 373
 374         return waitForShardResults(shardResultData, shardResults ->
 375                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 376         "       Failed to remove replica");
 377     }
 378
 379     @Override
 380     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
 381             final ChangeMemberVotingStatesForShardInput input) {
 382         final String shardName = input.getShardName();
 383         if (Strings.isNullOrEmpty(shardName)) {
 384             return newFailedRpcResultFuture("A valid shard name must be specified");
 385         }
 386
 387         DataStoreType dataStoreType = input.getDataStoreType();
 388         if (dataStoreType == null) {
 389             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 390         }
 391
 392         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 393         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 394             return newFailedRpcResultFuture("No member voting state input was specified");
 395         }
 396
 397         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 398                 memberVotingStates);
 399
 400         LOG.info("Change member voting states for shard {}: {}", shardName,
 401                 changeVotingStatus.getMeberVotingStatusMap());
 402
 403         final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
 404         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 405         Futures.addCallback(future, new FutureCallback<Success>() {
 406             @Override
 407             public void onSuccess(final Success success) {
 408                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 409                 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
 410             }
 411
 412             @Override
 413             public void onFailure(final Throwable failure) {
 414                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 415                         returnFuture, failure);
 416             }
 417         }, MoreExecutors.directExecutor());
 418
 419         return returnFuture;
 420     }
 421
 422     @Override
 423     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
 424             final ChangeMemberVotingStatesForAllShardsInput input) {
 425         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 426         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
 427             return newFailedRpcResultFuture("No member voting state input was specified");
 428         }
 429
 430         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 431         Function<String, Object> messageSupplier = shardName ->
 432                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 433
 434         LOG.info("Change member voting states for all shards");
 435
 436         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 437         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 438
 439         return waitForShardResults(shardResultData, shardResults ->
 440                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 441                 "Failed to change member voting states");
 442     }
 443
 444     @Override
 445     public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
 446             final FlipMemberVotingStatesForAllShardsInput input) {
 447         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 448         Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
 449
 450         LOG.info("Flip member voting states for all shards");
 451
 452         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 453         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 454
 455         return waitForShardResults(shardResultData, shardResults ->
 456                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 457                 "Failed to change member voting states");
 458     }
 459
 460     @Override
 461     public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
 462         final String shardName = input.getShardName();
 463         if (Strings.isNullOrEmpty(shardName)) {
 464             return newFailedRpcResultFuture("A valid shard name must be specified");
 465         }
 466
 467         DataStoreType dataStoreType = input.getDataStoreType();
 468         if (dataStoreType == null) {
 469             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 470         }
 471
 472         LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
 473
 474         final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
 475         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 476                 new GetShardRole(shardName));
 477         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 478             @Override
 479             public void onSuccess(final GetShardRoleReply reply) {
 480                 if (reply == null) {
 481                     returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 482                             "No Shard role present. Please retry..").build());
 483                     return;
 484                 }
 485                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 486                 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
 487                 if (reply.getRole() != null) {
 488                     builder.setRole(reply.getRole());
 489                 }
 490                 returnFuture.set(newSuccessfulResult(builder.build()));
 491             }
 492
 493             @Override
 494             public void onFailure(final Throwable failure) {
 495                 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
 496                         "Failed to get shard role.", failure).build());
 497             }
 498         }, MoreExecutors.directExecutor());
 499
 500         return returnFuture;
 501     }
 502
 503     @Override
 504     public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
 505             final GetPrefixShardRoleInput input) {
 506         final InstanceIdentifier<?> identifier = input.getShardPrefix();
 507         if (identifier == null) {
 508             return newFailedRpcResultFuture("A valid shard identifier must be specified");
 509         }
 510
 511         final DataStoreType dataStoreType = input.getDataStoreType();
 512         if (dataStoreType == null) {
 513             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 514         }
 515
 516         LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
 517
 518         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
 519         final String shardName = ClusterUtils.getCleanShardName(prefix);
 520         final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
 521         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
 522                 new GetShardRole(shardName));
 523         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
 524             @Override
 525             public void onSuccess(final GetShardRoleReply reply) {
 526                 if (reply == null) {
 527                     returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 528                             "No Shard role present. Please retry..").build());
 529                     return;
 530                 }
 531
 532                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
 533                 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
 534                 if (reply.getRole() != null) {
 535                     builder.setRole(reply.getRole());
 536                 }
 537                 returnFuture.set(newSuccessfulResult(builder.build()));
 538             }
 539
 540             @Override
 541             public void onFailure(final Throwable failure) {
 542                 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
 543                         "Failed to get shard role.", failure).build());
 544             }
 545         }, MoreExecutors.directExecutor());
 546
 547         return returnFuture;
 548     }
 549
 550     @Override
 551     public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
 552         LOG.debug("backupDatastore: {}", input);
 553
 554         if (Strings.isNullOrEmpty(input.getFilePath())) {
 555             return newFailedRpcResultFuture("A valid file path must be specified");
 556         }
 557
 558         final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
 559         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 560         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 561             @Override
 562             public void onSuccess(final List<DatastoreSnapshot> snapshots) {
 563                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 564             }
 565
 566             @Override
 567             public void onFailure(final Throwable failure) {
 568                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 569             }
 570         }, MoreExecutors.directExecutor());
 571
 572         return returnFuture;
 573     }
 574
 575     private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
 576             final List<MemberVotingState> memberVotingStatus) {
 577         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
 578         for (MemberVotingState memberStatus: memberVotingStatus) {
 579             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
 580         }
 581         return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
 582     }
 583
 584     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 585             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 586             final Function<List<ShardResult>, T> resultDataSupplier,
 587             final String failureLogMsgPrefix) {
 588         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 589         final List<ShardResult> shardResults = new ArrayList<>();
 590         for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
 591             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 592                 @Override
 593                 public void onSuccess(final Success result) {
 594                     synchronized (shardResults) {
 595                         ShardResultBuilder shardResult = entry.getValue();
 596                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 597                                 shardResult.getDataStoreType());
 598                         shardResults.add(shardResult.setSucceeded(true).build());
 599                         checkIfComplete();
 600                     }
 601                 }
 602
 603                 @Override
 604                 public void onFailure(final Throwable failure) {
 605                     synchronized (shardResults) {
 606                         ShardResultBuilder shardResult = entry.getValue();
 607                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 608                                 shardResult.getDataStoreType(), failure);
 609                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 610                                 Throwables.getRootCause(failure).getMessage()).build());
 611                         checkIfComplete();
 612                     }
 613                 }
 614
 615                 void checkIfComplete() {
 616                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 617                     if (shardResults.size() == shardResultData.size()) {
 618                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 619                     }
 620                 }
 621             }, MoreExecutors.directExecutor());
 622         }
 623         return returnFuture;
 624     }
 625
 626     private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
 627             final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 628             final Function<String, Object> messageSupplier) {
 629         ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
 630                 : operDataStore.getActorContext();
 631         Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
 632
 633         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
 634
 635         for (String shardName: allShardNames) {
 636             ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
 637                                                   SHARD_MGR_TIMEOUT);
 638             shardResultData.add(new SimpleEntry<>(future,
 639                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 640         }
 641     }
 642
 643     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
 644         Timeout timeout = SHARD_MGR_TIMEOUT;
 645         ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
 646         ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
 647
 648         return Futures.allAsList(configFuture, operFuture);
 649     }
 650
 651     private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
 652         ActorRef shardManager = dataStoreType == DataStoreType.Config
 653                 ? configDataStore.getActorContext().getShardManager()
 654                         : operDataStore.getActorContext().getShardManager();
 655         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 656     }
 657
 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.getActorContext().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 }