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