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