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.List;
  26 import java.util.Map.Entry;
  27 import java.util.Set;
  28 import java.util.concurrent.Future;
  29 import java.util.concurrent.TimeUnit;
  30 import org.apache.commons.lang3.SerializationUtils;
  31 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
  32 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
  33 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
  34 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshotList;
  35 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
  36 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
  37 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
  38 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
  39 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToNonvotingForAllShardsInput;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToVotingForAllShardsInput;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
  51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
  52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
  53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
  54 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  55 import org.opendaylight.yangtools.yang.common.RpcResult;
  56 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  57 import org.slf4j.Logger;
  58 import org.slf4j.LoggerFactory;
  59
  60 /**
  61  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
  62  *
  63  * @author Thomas Pantelis
  64  */
  65 public class ClusterAdminRpcService implements ClusterAdminService, AutoCloseable {
  66     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
  67
  68     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
  69
  70     private final DistributedDataStoreInterface configDataStore;
  71     private final DistributedDataStoreInterface operDataStore;
  72     private RpcRegistration<ClusterAdminService> rpcRegistration;
  73
  74     public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
  75             DistributedDataStoreInterface operDataStore) {
  76         this.configDataStore = configDataStore;
  77         this.operDataStore = operDataStore;
  78     }
  79
  80     public void start(RpcProviderRegistry rpcProviderRegistry) {
  81         LOG.debug("ClusterAdminRpcService starting");
  82
  83         rpcRegistration = rpcProviderRegistry.addRpcImplementation(ClusterAdminService.class, this);
  84     }
  85
  86     @Override
  87     public void close() {
  88         if(rpcRegistration != null) {
  89             rpcRegistration.close();
  90         }
  91     }
  92
  93     @Override
  94     public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
  95         final String shardName = input.getShardName();
  96         if(Strings.isNullOrEmpty(shardName)) {
  97             return newFailedRpcResultFuture("A valid shard name must be specified");
  98         }
  99
 100         DataStoreType dataStoreType = input.getDataStoreType();
 101         if(dataStoreType == null) {
 102             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 103         }
 104
 105         LOG.info("Adding replica for shard {}", shardName);
 106
 107         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 108         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
 109         Futures.addCallback(future, new FutureCallback<Success>() {
 110             @Override
 111             public void onSuccess(Success success) {
 112                 LOG.info("Successfully added replica for shard {}", shardName);
 113                 returnFuture.set(newSuccessfulResult());
 114             }
 115
 116             @Override
 117             public void onFailure(Throwable failure) {
 118                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
 119                         returnFuture, failure);
 120             }
 121         });
 122
 123         return returnFuture;
 124     }
 125
 126     @Override
 127     public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
 128         final String shardName = input.getShardName();
 129         if(Strings.isNullOrEmpty(shardName)) {
 130             return newFailedRpcResultFuture("A valid shard name must be specified");
 131         }
 132
 133         DataStoreType dataStoreType = input.getDataStoreType();
 134         if(dataStoreType == null) {
 135             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
 136         }
 137
 138         final String memberName = input.getMemberName();
 139         if(Strings.isNullOrEmpty(memberName)) {
 140             return newFailedRpcResultFuture("A valid member name must be specified");
 141         }
 142
 143         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
 144
 145         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 146         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
 147                 new RemoveShardReplica(shardName, memberName));
 148         Futures.addCallback(future, new FutureCallback<Success>() {
 149             @Override
 150             public void onSuccess(Success success) {
 151                 LOG.info("Successfully removed replica for shard {}", shardName);
 152                 returnFuture.set(newSuccessfulResult());
 153             }
 154
 155             @Override
 156             public void onFailure(Throwable failure) {
 157                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
 158                         returnFuture, failure);
 159             }
 160         });
 161
 162         return returnFuture;
 163     }
 164
 165     @Override
 166     public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
 167         LOG.info("Adding replicas for all shards");
 168
 169         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 170         Function<String, Object> messageSupplier = new Function<String, Object>() {
 171             @Override
 172             public Object apply(String shardName) {
 173                 return new AddShardReplica(shardName);
 174             }
 175         };
 176
 177         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 178         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 179
 180         return waitForShardResults(shardResultData, new Function<List<ShardResult>, AddReplicasForAllShardsOutput>() {
 181             @Override
 182             public AddReplicasForAllShardsOutput apply(List<ShardResult> shardResults) {
 183                 return new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build();
 184             }
 185         }, "Failed to add replica");
 186     }
 187
 188
 189     @Override
 190     public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
 191         LOG.info("Removing replicas for all shards");
 192
 193         final String memberName = input.getMemberName();
 194         if(Strings.isNullOrEmpty(memberName)) {
 195             return newFailedRpcResultFuture("A valid member name must be specified");
 196         }
 197
 198         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
 199         Function<String, Object> messageSupplier = new Function<String, Object>() {
 200             @Override
 201             public Object apply(String shardName) {
 202                 return new RemoveShardReplica(shardName, memberName);
 203             }
 204         };
 205
 206         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
 207         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
 208
 209         return waitForShardResults(shardResultData, new Function<List<ShardResult>, RemoveAllShardReplicasOutput>() {
 210             @Override
 211             public RemoveAllShardReplicasOutput apply(List<ShardResult> shardResults) {
 212                 return new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build();
 213             }
 214         }, "Failed to add replica");
 215     }
 216
 217     @Override
 218     public Future<RpcResult<Void>> convertMembersToVotingForAllShards(ConvertMembersToVotingForAllShardsInput input) {
 219         // TODO implement
 220         return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
 221                 "Not implemented yet").buildFuture();
 222     }
 223
 224     @Override
 225     public Future<RpcResult<Void>> convertMembersToNonvotingForAllShards(
 226             ConvertMembersToNonvotingForAllShardsInput input) {
 227         // TODO implement
 228         return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
 229                 "Not implemented yet").buildFuture();
 230     }
 231
 232     @Override
 233     public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
 234         LOG.debug("backupDatastore: {}", input);
 235
 236         if(Strings.isNullOrEmpty(input.getFilePath())) {
 237             return newFailedRpcResultFuture("A valid file path must be specified");
 238         }
 239
 240         final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
 241         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
 242         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
 243             @Override
 244             public void onSuccess(List<DatastoreSnapshot> snapshots) {
 245                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
 246             }
 247
 248             @Override
 249             public void onFailure(Throwable failure) {
 250                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
 251             }
 252         });
 253
 254         return returnFuture;
 255     }
 256
 257     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
 258             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
 259             final Function<List<ShardResult>, T> resultDataSupplier,
 260             final String failureLogMsgPrefix) {
 261         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
 262         final List<ShardResult> shardResults = new ArrayList<>();
 263         for(final Entry<ListenableFuture<Success>, ShardResultBuilder> entry: shardResultData) {
 264             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
 265                 @Override
 266                 public void onSuccess(Success result) {
 267                     synchronized(shardResults) {
 268                         ShardResultBuilder shardResult = entry.getValue();
 269                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
 270                                 shardResult.getDataStoreType());
 271                         shardResults.add(shardResult.setSucceeded(true).build());
 272                         checkIfComplete();
 273                     }
 274                 }
 275
 276                 @Override
 277                 public void onFailure(Throwable t) {
 278                     synchronized(shardResults) {
 279                         ShardResultBuilder shardResult = entry.getValue();
 280                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
 281                                 shardResult.getDataStoreType(), t);
 282                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
 283                                 Throwables.getRootCause(t).getMessage()).build());
 284                         checkIfComplete();
 285                     }
 286                 }
 287
 288                 void checkIfComplete() {
 289                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
 290                     if(shardResults.size() == shardResultData.size()) {
 291                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
 292                     }
 293                 }
 294             });
 295         }
 296         return returnFuture;
 297     }
 298
 299     private <T> void sendMessageToManagerForConfiguredShards(DataStoreType dataStoreType,
 300             List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
 301             Function<String, Object> messageSupplier) {
 302         ActorContext actorContext = dataStoreType == DataStoreType.Config ?
 303                 configDataStore.getActorContext() : operDataStore.getActorContext();
 304         Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
 305
 306         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
 307
 308         for(String shardName: allShardNames) {
 309             ListenableFuture<T> future = this.<T>ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
 310                     SHARD_MGR_TIMEOUT);
 311             shardResultData.add(new SimpleEntry<ListenableFuture<T>, ShardResultBuilder>(future,
 312                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
 313         }
 314     }
 315
 316     @SuppressWarnings("unchecked")
 317     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
 318         Timeout timeout = SHARD_MGR_TIMEOUT;
 319         ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
 320         ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
 321
 322         return Futures.allAsList(configFuture, operFuture);
 323     }
 324
 325     private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
 326         ActorRef shardManager = dataStoreType == DataStoreType.Config ?
 327                 configDataStore.getActorContext().getShardManager() : operDataStore.getActorContext().getShardManager();
 328         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
 329     }
 330
 331     private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
 332             SettableFuture<RpcResult<Void>> returnFuture) {
 333         try(FileOutputStream fos = new FileOutputStream(fileName)) {
 334             SerializationUtils.serialize(snapshots, fos);
 335
 336             returnFuture.set(newSuccessfulResult());
 337             LOG.info("Successfully backed up datastore to file {}", fileName);
 338         } catch(Exception e) {
 339             onDatastoreBackupFailure(fileName, returnFuture, e);
 340         }
 341     }
 342
 343     private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
 344             Throwable failure) {
 345         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
 346     }
 347
 348     private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
 349             Throwable failure) {
 350         LOG.error(msg, failure);
 351         returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
 352                 failure.getMessage())).build());
 353     }
 354
 355     private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
 356         final SettableFuture<T> returnFuture = SettableFuture.create();
 357
 358         @SuppressWarnings("unchecked")
 359         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
 360         askFuture.onComplete(new OnComplete<T>() {
 361             @Override
 362             public void onComplete(Throwable failure, T resp) {
 363                 if(failure != null) {
 364                     returnFuture.setException(failure);
 365                 } else {
 366                     returnFuture.set(resp);
 367                 }
 368             }
 369         }, configDataStore.getActorContext().getClientDispatcher());
 370
 371         return returnFuture;
 372     }
 373
 374     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
 375         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
 376     }
 377
 378     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
 379         return newFailedRpcResultBuilder(message, null);
 380     }
 381
 382     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
 383         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
 384     }
 385
 386     private static RpcResult<Void> newSuccessfulResult() {
 387         return newSuccessfulResult((Void)null);
 388     }
 389
 390     private static <T> RpcResult<T> newSuccessfulResult(T data) {
 391         return RpcResultBuilder.<T>success(data).build();
 392     }
 393 }