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