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.Strings;
16 import com.google.common.util.concurrent.FutureCallback;
17 import com.google.common.util.concurrent.Futures;
18 import com.google.common.util.concurrent.ListenableFuture;
19 import com.google.common.util.concurrent.SettableFuture;
20 import java.io.FileOutputStream;
21 import java.util.List;
22 import java.util.concurrent.Future;
23 import java.util.concurrent.TimeUnit;
24 import org.apache.commons.lang3.SerializationUtils;
25 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
26 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
27 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
28 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshotList;
29 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
30 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
31 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToNonvotingForAllShardsInput;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToVotingForAllShardsInput;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
39 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
40 import org.opendaylight.yangtools.yang.common.RpcResult;
41 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
42 import org.slf4j.Logger;
43 import org.slf4j.LoggerFactory;
46 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
48 * @author Thomas Pantelis
50 public class ClusterAdminRpcService implements ClusterAdminService, AutoCloseable {
51 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
53 private final DistributedDataStore configDataStore;
54 private final DistributedDataStore operDataStore;
55 private RpcRegistration<ClusterAdminService> rpcRegistration;
57 public ClusterAdminRpcService(DistributedDataStore configDataStore, DistributedDataStore operDataStore) {
58 this.configDataStore = configDataStore;
59 this.operDataStore = operDataStore;
62 public void start(RpcProviderRegistry rpcProviderRegistry) {
63 LOG.debug("ClusterAdminRpcService starting");
65 rpcRegistration = rpcProviderRegistry.addRpcImplementation(ClusterAdminService.class, this);
70 if(rpcRegistration != null) {
71 rpcRegistration.close();
76 public Future<RpcResult<Void>> addShardReplica(final AddShardReplicaInput input) {
77 final String shardName = input.getShardName();
78 if(Strings.isNullOrEmpty(shardName)) {
79 return newFailedRpcResultBuilder("A valid shard name must be specified").buildFuture();
82 DataStoreType dataStoreType = input.getDataStoreType();
83 if(dataStoreType == null) {
84 return newFailedRpcResultBuilder("A valid DataStoreType must be specified").buildFuture();
87 LOG.info("Adding replica for shard {}", shardName);
89 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
90 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
91 Futures.addCallback(future, new FutureCallback<Success>() {
93 public void onSuccess(Success success) {
94 LOG.info("Successfully added replica for shard {}", shardName);
95 returnFuture.set(newSuccessfulResult());
99 public void onFailure(Throwable failure) {
100 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
101 returnFuture, failure);
109 public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
111 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
112 "Not implemented yet").buildFuture();
116 public Future<RpcResult<Void>> addReplicasForAllShards() {
118 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
119 "Not implemented yet").buildFuture();
123 public Future<RpcResult<Void>> removeAllShardReplicas() {
125 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
126 "Not implemented yet").buildFuture();
130 public Future<RpcResult<Void>> convertMembersToVotingForAllShards(ConvertMembersToVotingForAllShardsInput input) {
132 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
133 "Not implemented yet").buildFuture();
137 public Future<RpcResult<Void>> convertMembersToNonvotingForAllShards(
138 ConvertMembersToNonvotingForAllShardsInput input) {
140 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
141 "Not implemented yet").buildFuture();
145 public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
146 LOG.debug("backupDatastore: {}", input);
148 if(Strings.isNullOrEmpty(input.getFilePath())) {
149 return newFailedRpcResultBuilder("A valid file path must be specified").buildFuture();
152 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
153 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
154 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
156 public void onSuccess(List<DatastoreSnapshot> snapshots) {
157 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
161 public void onFailure(Throwable failure) {
162 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
169 @SuppressWarnings("unchecked")
170 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(Object message) {
171 Timeout timeout = new Timeout(1, TimeUnit.MINUTES);
172 ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
173 ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
175 return Futures.allAsList(configFuture, operFuture);
178 private <T> ListenableFuture<T> sendMessageToShardManager(DataStoreType dataStoreType, Object message) {
179 ActorRef shardManager = dataStoreType == DataStoreType.Config ?
180 configDataStore.getActorContext().getShardManager() : operDataStore.getActorContext().getShardManager();
181 return ask(shardManager, message, new Timeout(1, TimeUnit.MINUTES));
184 private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
185 SettableFuture<RpcResult<Void>> returnFuture) {
186 try(FileOutputStream fos = new FileOutputStream(fileName)) {
187 SerializationUtils.serialize(snapshots, fos);
189 returnFuture.set(newSuccessfulResult());
190 LOG.info("Successfully backed up datastore to file {}", fileName);
191 } catch(Exception e) {
192 onDatastoreBackupFailure(fileName, returnFuture, e);
196 private static void onDatastoreBackupFailure(String fileName, SettableFuture<RpcResult<Void>> returnFuture,
198 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
201 private static void onMessageFailure(String msg, final SettableFuture<RpcResult<Void>> returnFuture,
203 LOG.error(msg, failure);
204 returnFuture.set(newFailedRpcResultBuilder(String.format("%s: %s", msg, failure.getMessage())).build());
207 private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
208 final SettableFuture<T> returnFuture = SettableFuture.create();
210 @SuppressWarnings("unchecked")
211 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
212 askFuture.onComplete(new OnComplete<T>() {
214 public void onComplete(Throwable failure, T resp) {
215 if(failure != null) {
216 returnFuture.setException(failure);
218 returnFuture.set(resp);
221 }, configDataStore.getActorContext().getClientDispatcher());
226 private static RpcResultBuilder<Void> newFailedRpcResultBuilder(String message) {
227 return newFailedRpcResultBuilder(message, null);
230 private static RpcResultBuilder<Void> newFailedRpcResultBuilder(String message, Throwable cause) {
231 return RpcResultBuilder.<Void>failed().withError(ErrorType.RPC, message, cause);
234 private static RpcResult<Void> newSuccessfulResult() {
235 return RpcResultBuilder.<Void>success().build();