2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore.admin;
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;
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.access.concepts.MemberName;
32 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
33 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
34 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
35 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshotList;
36 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
37 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
38 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
39 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
40 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToNonvotingForAllShardsInput;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToVotingForAllShardsInput;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
55 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
56 import org.opendaylight.yangtools.yang.common.RpcResult;
57 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
58 import org.slf4j.Logger;
59 import org.slf4j.LoggerFactory;
62 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
64 * @author Thomas Pantelis
66 public class ClusterAdminRpcService implements ClusterAdminService, AutoCloseable {
67 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
69 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
71 private final DistributedDataStoreInterface configDataStore;
72 private final DistributedDataStoreInterface operDataStore;
73 private RpcRegistration<ClusterAdminService> rpcRegistration;
75 public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
76 DistributedDataStoreInterface operDataStore) {
77 this.configDataStore = configDataStore;
78 this.operDataStore = operDataStore;
81 public void start(RpcProviderRegistry rpcProviderRegistry) {
82 LOG.debug("ClusterAdminRpcService starting");
84 rpcRegistration = rpcProviderRegistry.addRpcImplementation(ClusterAdminService.class, this);
89 if(rpcRegistration != null) {
90 rpcRegistration.close();
95 public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
96 final String shardName = input.getShardName();
97 if(Strings.isNullOrEmpty(shardName)) {
98 return newFailedRpcResultFuture("A valid shard name must be specified");
101 DataStoreType dataStoreType = input.getDataStoreType();
102 if(dataStoreType == null) {
103 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
106 LOG.info("Adding replica for shard {}", shardName);
108 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
109 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
110 Futures.addCallback(future, new FutureCallback<Success>() {
112 public void onSuccess(Success success) {
113 LOG.info("Successfully added replica for shard {}", shardName);
114 returnFuture.set(newSuccessfulResult());
118 public void onFailure(Throwable failure) {
119 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
120 returnFuture, failure);
128 public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
129 final String shardName = input.getShardName();
130 if(Strings.isNullOrEmpty(shardName)) {
131 return newFailedRpcResultFuture("A valid shard name must be specified");
134 DataStoreType dataStoreType = input.getDataStoreType();
135 if(dataStoreType == null) {
136 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
139 final String memberName = input.getMemberName();
140 if(Strings.isNullOrEmpty(memberName)) {
141 return newFailedRpcResultFuture("A valid member name must be specified");
144 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
146 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
147 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
148 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
149 Futures.addCallback(future, new FutureCallback<Success>() {
151 public void onSuccess(Success success) {
152 LOG.info("Successfully removed replica for shard {}", shardName);
153 returnFuture.set(newSuccessfulResult());
157 public void onFailure(Throwable failure) {
158 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
159 returnFuture, failure);
167 public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
168 LOG.info("Adding replicas for all shards");
170 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
171 Function<String, Object> messageSupplier = new Function<String, Object>() {
173 public Object apply(String shardName) {
174 return new AddShardReplica(shardName);
178 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
179 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
181 return waitForShardResults(shardResultData, new Function<List<ShardResult>, AddReplicasForAllShardsOutput>() {
183 public AddReplicasForAllShardsOutput apply(List<ShardResult> shardResults) {
184 return new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build();
186 }, "Failed to add replica");
191 public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
192 LOG.info("Removing replicas for all shards");
194 final String memberName = input.getMemberName();
195 if(Strings.isNullOrEmpty(memberName)) {
196 return newFailedRpcResultFuture("A valid member name must be specified");
199 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
200 Function<String, Object> messageSupplier = new Function<String, Object>() {
202 public Object apply(String shardName) {
203 return new RemoveShardReplica(shardName, MemberName.forName(memberName));
207 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
208 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
210 return waitForShardResults(shardResultData, new Function<List<ShardResult>, RemoveAllShardReplicasOutput>() {
212 public RemoveAllShardReplicasOutput apply(List<ShardResult> shardResults) {
213 return new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build();
215 }, "Failed to add replica");
219 public Future<RpcResult<Void>> convertMembersToVotingForAllShards(ConvertMembersToVotingForAllShardsInput input) {
221 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
222 "Not implemented yet").buildFuture();
226 public Future<RpcResult<Void>> convertMembersToNonvotingForAllShards(
227 ConvertMembersToNonvotingForAllShardsInput input) {
229 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
230 "Not implemented yet").buildFuture();
234 public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
235 LOG.debug("backupDatastore: {}", input);
237 if(Strings.isNullOrEmpty(input.getFilePath())) {
238 return newFailedRpcResultFuture("A valid file path must be specified");
241 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
242 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
243 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
245 public void onSuccess(List<DatastoreSnapshot> snapshots) {
246 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
250 public void onFailure(Throwable failure) {
251 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
258 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
259 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
260 final Function<List<ShardResult>, T> resultDataSupplier,
261 final String failureLogMsgPrefix) {
262 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
263 final List<ShardResult> shardResults = new ArrayList<>();
264 for(final Entry<ListenableFuture<Success>, ShardResultBuilder> entry: shardResultData) {
265 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
267 public void onSuccess(Success result) {
268 synchronized(shardResults) {
269 ShardResultBuilder shardResult = entry.getValue();
270 LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
271 shardResult.getDataStoreType());
272 shardResults.add(shardResult.setSucceeded(true).build());
278 public void onFailure(Throwable t) {
279 synchronized(shardResults) {
280 ShardResultBuilder shardResult = entry.getValue();
281 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
282 shardResult.getDataStoreType(), t);
283 shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
284 Throwables.getRootCause(t).getMessage()).build());
289 void checkIfComplete() {
290 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
291 if(shardResults.size() == shardResultData.size()) {
292 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
300 private <T> void sendMessageToManagerForConfiguredShards(DataStoreType dataStoreType,
301 List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
302 Function<String, Object> messageSupplier) {
303 ActorContext actorContext = dataStoreType == DataStoreType.Config ?
304 configDataStore.getActorContext() : operDataStore.getActorContext();
305 Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
307 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
309 for(String shardName: allShardNames) {
310 ListenableFuture<T> future = this.<T>ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
312 shardResultData.add(new SimpleEntry<ListenableFuture<T>, ShardResultBuilder>(future,
313 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
317 @SuppressWarnings("unchecked")
318 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
319 Timeout timeout = SHARD_MGR_TIMEOUT;
320 ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
321 ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
323 return Futures.allAsList(configFuture, operFuture);
326 private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
327 ActorRef shardManager = dataStoreType == DataStoreType.Config ?
328 configDataStore.getActorContext().getShardManager() : operDataStore.getActorContext().getShardManager();
329 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
332 private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
333 SettableFuture<RpcResult<Void>> returnFuture) {
334 try(FileOutputStream fos = new FileOutputStream(fileName)) {
335 SerializationUtils.serialize(snapshots, fos);
337 returnFuture.set(newSuccessfulResult());
338 LOG.info("Successfully backed up datastore to file {}", fileName);
339 } catch(Exception e) {
340 onDatastoreBackupFailure(fileName, returnFuture, e);
344 private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
346 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
349 private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
351 LOG.error(msg, failure);
352 returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
353 failure.getMessage())).build());
356 private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
357 final SettableFuture<T> returnFuture = SettableFuture.create();
359 @SuppressWarnings("unchecked")
360 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
361 askFuture.onComplete(new OnComplete<T>() {
363 public void onComplete(Throwable failure, T resp) {
364 if(failure != null) {
365 returnFuture.setException(failure);
367 returnFuture.set(resp);
370 }, configDataStore.getActorContext().getClientDispatcher());
375 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
376 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
379 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
380 return newFailedRpcResultBuilder(message, null);
383 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
384 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
387 private static RpcResult<Void> newSuccessfulResult() {
388 return newSuccessfulResult((Void)null);
391 private static <T> RpcResult<T> newSuccessfulResult(T data) {
392 return RpcResultBuilder.<T>success(data).build();