opendaylight/md-sal/sal-cluster-admin/src/main/java/org/opendaylight/controller/cluster/datastore/admin/ClusterAdminRpcService.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore.admin;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.Status.Success;
  12 import akka.dispatch.OnComplete;
  13 import akka.pattern.Patterns;
  14 import akka.util.Timeout;
  15 import com.google.common.base.Function;
  16 import com.google.common.base.Strings;
  17 import com.google.common.base.Throwables;
  18 import com.google.common.util.concurrent.FutureCallback;
  19 import com.google.common.util.concurrent.Futures;
  20 import com.google.common.util.concurrent.ListenableFuture;
  21 import com.google.common.util.concurrent.SettableFuture;
  22 import java.io.FileOutputStream;
  23 import java.util.AbstractMap.SimpleEntry;
  24 import java.util.ArrayList;
  25 import java.util.HashMap;
  26 import java.util.List;
  27 import java.util.Map;
  28 import java.util.Map.Entry;
  29 import java.util.Set;
  30 import java.util.concurrent.Future;
  31 import java.util.concurrent.TimeUnit;
  32 import org.apache.commons.lang3.SerializationUtils;
  33 import org.opendaylight.controller.cluster.access.concepts.MemberName;
  34 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  35 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  36 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
  37 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
  38 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshotList;
  39 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
  40 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  41 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  42 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
  54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
  55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
  56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
  57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
  58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
  59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
  60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
  61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
  62 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  63 import org.opendaylight.yangtools.yang.common.RpcResult;
  64 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  65 import org.slf4j.Logger;
  66 import org.slf4j.LoggerFactory;
  67
  68 /**
  69  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
  70  *
  71  * @author Thomas Pantelis
  72  */
  73 public class ClusterAdminRpcService implements ClusterAdminService {
  74     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
  75
  76     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
  77
  78     private final DistributedDataStoreInterface configDataStore;
  79     private final DistributedDataStoreInterface operDataStore;
  80
  81     public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
  82             DistributedDataStoreInterface operDataStore) {
  83         this.configDataStore = configDataStore;
  84         this.operDataStore = operDataStore;
  85     }
  86
  87     @Override
  88     public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
  89         final String shardName = input.getShardName();
  90         if(Strings.isNullOrEmpty(shardName)) {
  91             return newFailedRpcResultFuture("A valid shard name must be specified");
  92         }
  93
  94         DataStoreType dataStoreType = input.getDataStoreType();
  95         if(dataStoreType == null) {
  96             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
  97         }
  98
  99         LOG.info("Adding replica for shard {}", shardName);
 100
 101         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 102         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 103         Futures.addCallback(future, new FutureCallback<Success>() {
 104             @Override
 105             public void onSuccess(Success success) {
 106                 LOG.info("Successfully added replica for shard {}", shardName);
 107                 returnFuture.set(newSuccessfulResult());
 108             }
 109
 110             @Override
 111             public void onFailure(Throwable failure) {
 112                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 113                         returnFuture, failure);
 114             }
 115         });
 116
 117         return returnFuture;
 118     }
 119
 120     @Override
 121     public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
 122         final String shardName = input.getShardName();
 123         if(Strings.isNullOrEmpty(shardName)) {
 124             return newFailedRpcResultFuture("A valid shard name must be specified");
 125         }
 126
 127         DataStoreType dataStoreType = input.getDataStoreType();
 128         if(dataStoreType == null) {
 129             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 130         }
 131
 132         final String memberName = input.getMemberName();
 133         if(Strings.isNullOrEmpty(memberName)) {
 134             return newFailedRpcResultFuture("A valid member name must be specified");
 135         }
 136
 137         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 138
 139         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 140         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 141                 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
 142         Futures.addCallback(future, new FutureCallback<Success>() {
 143             @Override
 144             public void onSuccess(Success success) {
 145                 LOG.info("Successfully removed replica for shard {}", shardName);
 146                 returnFuture.set(newSuccessfulResult());
 147             }
 148
 149             @Override
 150             public void onFailure(Throwable failure) {
 151                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 152                         returnFuture, failure);
 153             }
 154         });
 155
 156         return returnFuture;
 157     }
 158
 159     @Override
 160     public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
 161         LOG.info("Adding replicas for all shards");
 162
 163         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 164         Function<String, Object> messageSupplier = shardName -> new AddShardReplica(shardName);
 165
 166         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 167         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 168
 169         return waitForShardResults(shardResultData, shardResults ->
 170                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 171                 "Failed to add replica");
 172     }
 173
 174
 175     @Override
 176     public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
 177         LOG.info("Removing replicas for all shards");
 178
 179         final String memberName = input.getMemberName();
 180         if(Strings.isNullOrEmpty(memberName)) {
 181             return newFailedRpcResultFuture("A valid member name must be specified");
 182         }
 183
 184         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 185         Function<String, Object> messageSupplier = shardName ->
 186                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
 187
 188         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 189         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 190
 191         return waitForShardResults(shardResultData, shardResults ->
 192                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
 193         "       Failed to remove replica");
 194     }
 195
 196     @Override
 197     public Future<RpcResult<Void>> changeMemberVotingStatesForShard(ChangeMemberVotingStatesForShardInput input) {
 198         final String shardName = input.getShardName();
 199         if(Strings.isNullOrEmpty(shardName)) {
 200             return newFailedRpcResultFuture("A valid shard name must be specified");
 201         }
 202
 203         DataStoreType dataStoreType = input.getDataStoreType();
 204         if(dataStoreType == null) {
 205             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 206         }
 207
 208         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 209         if(memberVotingStates == null || memberVotingStates.isEmpty()) {
 210             return newFailedRpcResultFuture("No member voting state input was specified");
 211         }
 212
 213         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
 214                 memberVotingStates);
 215
 216         LOG.info("Change member voting states for shard {}: {}", shardName,
 217                 changeVotingStatus.getMeberVotingStatusMap());
 218
 219         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 220         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
 221         Futures.addCallback(future, new FutureCallback<Success>() {
 222             @Override
 223             public void onSuccess(Success success) {
 224                 LOG.info("Successfully changed member voting states for shard {}", shardName);
 225                 returnFuture.set(newSuccessfulResult());
 226             }
 227
 228             @Override
 229             public void onFailure(Throwable failure) {
 230                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
 231                         returnFuture, failure);
 232             }
 233         });
 234
 235         return returnFuture;
 236     }
 237
 238     @Override
 239     public Future<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
 240             final ChangeMemberVotingStatesForAllShardsInput input) {
 241         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
 242         if(memberVotingStates == null || memberVotingStates.isEmpty()) {
 243             return newFailedRpcResultFuture("No member voting state input was specified");
 244         }
 245
 246         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 247         Function<String, Object> messageSupplier = shardName ->
 248                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
 249
 250         LOG.info("Change member voting states for all shards");
 251
 252         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 253         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 254
 255         return waitForShardResults(shardResultData, shardResults ->
 256                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 257                 "Failed to change member voting states");
 258     }
 259
 260     @Override
 261     public Future<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards() {
 262         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 263         Function<String, Object> messageSupplier = shardName ->
 264                 new FlipShardMembersVotingStatus(shardName);
 265
 266         LOG.info("Flip member voting states for all shards");
 267
 268         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 269         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 270
 271         return waitForShardResults(shardResultData, shardResults ->
 272                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
 273                 "Failed to change member voting states");
 274     }
 275
 276     @Override
 277     public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
 278         LOG.debug("backupDatastore: {}", input);
 279
 280         if(Strings.isNullOrEmpty(input.getFilePath())) {
 281             return newFailedRpcResultFuture("A valid file path must be specified");
 282         }
 283
 284         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 285         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 286         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 287             @Override
 288             public void onSuccess(List<DatastoreSnapshot> snapshots) {
 289                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 290             }
 291
 292             @Override
 293             public void onFailure(Throwable failure) {
 294                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 295             }
 296         });
 297
 298         return returnFuture;
 299     }
 300
 301     private ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
 302             List<MemberVotingState> memberVotingStatus) {
 303         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
 304         for(MemberVotingState memberStatus: memberVotingStatus) {
 305             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
 306         }
 307
 308         ChangeShardMembersVotingStatus changeVotingStatus = new ChangeShardMembersVotingStatus(shardName,
 309                 serverVotingStatusMap);
 310         return changeVotingStatus;
 311     }
 312
 313     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 314             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 315             final Function<List<ShardResult>, T> resultDataSupplier,
 316             final String failureLogMsgPrefix) {
 317         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 318         final List<ShardResult> shardResults = new ArrayList<>();
 319         for(final Entry<ListenableFuture<Success>, ShardResultBuilder> entry: shardResultData) {
 320             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 321                 @Override
 322                 public void onSuccess(Success result) {
 323                     synchronized(shardResults) {
 324                         ShardResultBuilder shardResult = entry.getValue();
 325                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 326                                 shardResult.getDataStoreType());
 327                         shardResults.add(shardResult.setSucceeded(true).build());
 328                         checkIfComplete();
 329                     }
 330                 }
 331
 332                 @Override
 333                 public void onFailure(Throwable t) {
 334                     synchronized(shardResults) {
 335                         ShardResultBuilder shardResult = entry.getValue();
 336                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 337                                 shardResult.getDataStoreType(), t);
 338                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 339                                 Throwables.getRootCause(t).getMessage()).build());
 340                         checkIfComplete();
 341                     }
 342                 }
 343
 344                 void checkIfComplete() {
 345                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 346                     if(shardResults.size() == shardResultData.size()) {
 347                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 348                     }
 349                 }
 350             });
 351         }
 352         return returnFuture;
 353     }
 354
 355     private <T> void sendMessageToManagerForConfiguredShards(DataStoreType dataStoreType,
 356             List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 357             Function<String, Object> messageSupplier) {
 358         ActorContext actorContext = dataStoreType == DataStoreType.Config ?
 359                 configDataStore.getActorContext() : operDataStore.getActorContext();
 360         Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
 361
 362         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
 363
 364         for(String shardName: allShardNames) {
 365             ListenableFuture<T> future = this.<T>ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
 366                     SHARD_MGR_TIMEOUT);
 367             shardResultData.add(new SimpleEntry<>(future,
 368                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 369         }
 370     }
 371
 372     @SuppressWarnings("unchecked")
 373     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
 374         Timeout timeout = SHARD_MGR_TIMEOUT;
 375         ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
 376         ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
 377
 378         return Futures.allAsList(configFuture, operFuture);
 379     }
 380
 381     private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
 382         ActorRef shardManager = dataStoreType == DataStoreType.Config ?
 383                 configDataStore.getActorContext().getShardManager() : operDataStore.getActorContext().getShardManager();
 384         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 385     }
 386
 387     private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
 388             SettableFuture<RpcResult<Void>> returnFuture) {
 389         try(FileOutputStream fos = new FileOutputStream(fileName)) {
 390             SerializationUtils.serialize(snapshots, fos);
 391
 392             returnFuture.set(newSuccessfulResult());
 393             LOG.info("Successfully backed up datastore to file {}", fileName);
 394         } catch(Exception e) {
 395             onDatastoreBackupFailure(fileName, returnFuture, e);
 396         }
 397     }
 398
 399     private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
 400             Throwable failure) {
 401         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 402     }
 403
 404     private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
 405             Throwable failure) {
 406         LOG.error(msg, failure);
 407         returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 408                 failure.getMessage())).build());
 409     }
 410
 411     private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
 412         final SettableFuture<T> returnFuture = SettableFuture.create();
 413
 414         @SuppressWarnings("unchecked")
 415         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 416         askFuture.onComplete(new OnComplete<T>() {
 417             @Override
 418             public void onComplete(Throwable failure, T resp) {
 419                 if(failure != null) {
 420                     returnFuture.setException(failure);
 421                 } else {
 422                     returnFuture.set(resp);
 423                 }
 424             }
 425         }, configDataStore.getActorContext().getClientDispatcher());
 426
 427         return returnFuture;
 428     }
 429
 430     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
 431         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 432     }
 433
 434     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
 435         return newFailedRpcResultBuilder(message, null);
 436     }
 437
 438     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
 439         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 440     }
 441
 442     private static RpcResult<Void> newSuccessfulResult() {
 443         return newSuccessfulResult((Void)null);
 444     }
 445
 446     private static <T> RpcResult<T> newSuccessfulResult(T data) {
 447         return RpcResultBuilder.<T>success(data).build();
 448     }
 449 }