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.HashMap;
26 import java.util.List;
28 import java.util.Map.Entry;
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;
69 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
71 * @author Thomas Pantelis
73 public class ClusterAdminRpcService implements ClusterAdminService {
74 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
76 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
78 private final DistributedDataStoreInterface configDataStore;
79 private final DistributedDataStoreInterface operDataStore;
81 public ClusterAdminRpcService(DistributedDataStoreInterface configDataStore,
82 DistributedDataStoreInterface operDataStore) {
83 this.configDataStore = configDataStore;
84 this.operDataStore = operDataStore;
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");
94 DataStoreType dataStoreType = input.getDataStoreType();
95 if(dataStoreType == null) {
96 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
99 LOG.info("Adding replica for shard {}", shardName);
101 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
102 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
103 Futures.addCallback(future, new FutureCallback<Success>() {
105 public void onSuccess(Success success) {
106 LOG.info("Successfully added replica for shard {}", shardName);
107 returnFuture.set(newSuccessfulResult());
111 public void onFailure(Throwable failure) {
112 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
113 returnFuture, failure);
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");
127 DataStoreType dataStoreType = input.getDataStoreType();
128 if(dataStoreType == null) {
129 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
132 final String memberName = input.getMemberName();
133 if(Strings.isNullOrEmpty(memberName)) {
134 return newFailedRpcResultFuture("A valid member name must be specified");
137 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
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>() {
144 public void onSuccess(Success success) {
145 LOG.info("Successfully removed replica for shard {}", shardName);
146 returnFuture.set(newSuccessfulResult());
150 public void onFailure(Throwable failure) {
151 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
152 returnFuture, failure);
160 public Future<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards() {
161 LOG.info("Adding replicas for all shards");
163 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
164 Function<String, Object> messageSupplier = shardName -> new AddShardReplica(shardName);
166 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
167 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
169 return waitForShardResults(shardResultData, shardResults ->
170 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
171 "Failed to add replica");
176 public Future<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(RemoveAllShardReplicasInput input) {
177 LOG.info("Removing replicas for all shards");
179 final String memberName = input.getMemberName();
180 if(Strings.isNullOrEmpty(memberName)) {
181 return newFailedRpcResultFuture("A valid member name must be specified");
184 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
185 Function<String, Object> messageSupplier = shardName ->
186 new RemoveShardReplica(shardName, MemberName.forName(memberName));
188 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
189 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
191 return waitForShardResults(shardResultData, shardResults ->
192 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
193 " Failed to remove replica");
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");
203 DataStoreType dataStoreType = input.getDataStoreType();
204 if(dataStoreType == null) {
205 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
208 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
209 if(memberVotingStates == null || memberVotingStates.isEmpty()) {
210 return newFailedRpcResultFuture("No member voting state input was specified");
213 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
216 LOG.info("Change member voting states for shard {}: {}", shardName,
217 changeVotingStatus.getMeberVotingStatusMap());
219 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
220 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
221 Futures.addCallback(future, new FutureCallback<Success>() {
223 public void onSuccess(Success success) {
224 LOG.info("Successfully changed member voting states for shard {}", shardName);
225 returnFuture.set(newSuccessfulResult());
229 public void onFailure(Throwable failure) {
230 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
231 returnFuture, failure);
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");
246 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
247 Function<String, Object> messageSupplier = shardName ->
248 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
250 LOG.info("Change member voting states for all shards");
252 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
253 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
255 return waitForShardResults(shardResultData, shardResults ->
256 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
257 "Failed to change member voting states");
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);
266 LOG.info("Flip member voting states for all shards");
268 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
269 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
271 return waitForShardResults(shardResultData, shardResults ->
272 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
273 "Failed to change member voting states");
277 public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
278 LOG.debug("backupDatastore: {}", input);
280 if(Strings.isNullOrEmpty(input.getFilePath())) {
281 return newFailedRpcResultFuture("A valid file path must be specified");
284 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
285 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
286 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
288 public void onSuccess(List<DatastoreSnapshot> snapshots) {
289 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
293 public void onFailure(Throwable failure) {
294 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
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());
308 ChangeShardMembersVotingStatus changeVotingStatus = new ChangeShardMembersVotingStatus(shardName,
309 serverVotingStatusMap);
310 return changeVotingStatus;
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>() {
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());
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());
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)));
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();
362 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
364 for(String shardName: allShardNames) {
365 ListenableFuture<T> future = this.<T>ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
367 shardResultData.add(new SimpleEntry<>(future,
368 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
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);
378 return Futures.allAsList(configFuture, operFuture);
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);
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);
392 returnFuture.set(newSuccessfulResult());
393 LOG.info("Successfully backed up datastore to file {}", fileName);
394 } catch(Exception e) {
395 onDatastoreBackupFailure(fileName, returnFuture, e);
399 private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
401 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
404 private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
406 LOG.error(msg, failure);
407 returnFuture.set(ClusterAdminRpcService.<Void>newFailedRpcResultBuilder(String.format("%s: %s", msg,
408 failure.getMessage())).build());
411 private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
412 final SettableFuture<T> returnFuture = SettableFuture.create();
414 @SuppressWarnings("unchecked")
415 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
416 askFuture.onComplete(new OnComplete<T>() {
418 public void onComplete(Throwable failure, T resp) {
419 if(failure != null) {
420 returnFuture.setException(failure);
422 returnFuture.set(resp);
425 }, configDataStore.getActorContext().getClientDispatcher());
430 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(String message) {
431 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
434 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message) {
435 return newFailedRpcResultBuilder(message, null);
438 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(String message, Throwable cause) {
439 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
442 private static RpcResult<Void> newSuccessfulResult() {
443 return newSuccessfulResult((Void)null);
446 private static <T> RpcResult<T> newSuccessfulResult(T data) {
447 return RpcResultBuilder.<T>success(data).build();