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.dispatch.OnComplete;
12 import akka.pattern.Patterns;
13 import akka.util.Timeout;
14 import com.google.common.base.Strings;
15 import com.google.common.util.concurrent.FutureCallback;
16 import com.google.common.util.concurrent.Futures;
17 import com.google.common.util.concurrent.ListenableFuture;
18 import com.google.common.util.concurrent.SettableFuture;
19 import java.io.FileOutputStream;
20 import java.util.List;
21 import java.util.concurrent.Future;
22 import java.util.concurrent.TimeUnit;
23 import org.apache.commons.lang3.SerializationUtils;
24 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
25 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
26 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshotList;
27 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
28 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
29 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToNonvotingForAllShardsInput;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ConvertMembersToVotingForAllShardsInput;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
36 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
37 import org.opendaylight.yangtools.yang.common.RpcResult;
38 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
39 import org.slf4j.Logger;
40 import org.slf4j.LoggerFactory;
43 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
45 * @author Thomas Pantelis
47 public class ClusterAdminRpcService implements ClusterAdminService, AutoCloseable {
48 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
50 private final DistributedDataStore configDataStore;
51 private final DistributedDataStore operDataStore;
52 private RpcRegistration<ClusterAdminService> rpcRegistration;
54 public ClusterAdminRpcService(DistributedDataStore configDataStore, DistributedDataStore operDataStore) {
55 this.configDataStore = configDataStore;
56 this.operDataStore = operDataStore;
59 public void start(RpcProviderRegistry rpcProviderRegistry) {
60 LOG.debug("ClusterAdminRpcService starting");
62 rpcRegistration = rpcProviderRegistry.addRpcImplementation(ClusterAdminService.class, this);
67 if(rpcRegistration != null) {
68 rpcRegistration.close();
73 public Future<RpcResult<Void>> addShardReplica(AddShardReplicaInput input) {
75 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
76 "Not implemented yet").buildFuture();
80 public Future<RpcResult<Void>> removeShardReplica(RemoveShardReplicaInput input) {
82 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
83 "Not implemented yet").buildFuture();
87 public Future<RpcResult<Void>> addReplicasForAllShards() {
89 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
90 "Not implemented yet").buildFuture();
94 public Future<RpcResult<Void>> removeAllShardReplicas() {
96 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
97 "Not implemented yet").buildFuture();
101 public Future<RpcResult<Void>> convertMembersToVotingForAllShards(ConvertMembersToVotingForAllShardsInput input) {
103 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
104 "Not implemented yet").buildFuture();
108 public Future<RpcResult<Void>> convertMembersToNonvotingForAllShards(
109 ConvertMembersToNonvotingForAllShardsInput input) {
111 return RpcResultBuilder.<Void>failed().withError(ErrorType.APPLICATION, "operation-not-supported",
112 "Not implemented yet").buildFuture();
115 @SuppressWarnings("unchecked")
117 public Future<RpcResult<Void>> backupDatastore(final BackupDatastoreInput input) {
118 LOG.debug("backupDatastore: {}", input);
120 if(Strings.isNullOrEmpty(input.getFilePath())) {
121 return newFailedRpcResultBuilder("A valid file path must be specified").buildFuture();
124 Timeout timeout = new Timeout(1, TimeUnit.MINUTES);
125 ListenableFuture<DatastoreSnapshot> configFuture = ask(configDataStore.getActorContext().getShardManager(),
126 GetSnapshot.INSTANCE, timeout);
127 ListenableFuture<DatastoreSnapshot> operFuture = ask(operDataStore.getActorContext().getShardManager(),
128 GetSnapshot.INSTANCE, timeout);
130 final SettableFuture<RpcResult<Void>> returnFuture = SettableFuture.create();
131 Futures.addCallback(Futures.allAsList(configFuture, operFuture), new FutureCallback<List<DatastoreSnapshot>>() {
133 public void onSuccess(List<DatastoreSnapshot> snapshots) {
134 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
138 public void onFailure(Throwable failure) {
139 onDatastoreBackupFilure(input.getFilePath(), returnFuture, failure);
146 private static void saveSnapshotsToFile(DatastoreSnapshotList snapshots, String fileName,
147 SettableFuture<RpcResult<Void>> returnFuture) {
148 try(FileOutputStream fos = new FileOutputStream(fileName)) {
149 SerializationUtils.serialize(snapshots, fos);
151 returnFuture.set(newSuccessfulResult());
152 LOG.info("Successfully backed up datastore to file {}", fileName);
153 } catch(Exception e) {
154 onDatastoreBackupFilure(fileName, returnFuture, e);
158 private static void onDatastoreBackupFilure(String fileName, final SettableFuture<RpcResult<Void>> returnFuture,
160 String msg = String.format("Failed to back up datastore to file %s", fileName);
161 LOG.error(msg, failure);
162 returnFuture.set(newFailedRpcResultBuilder(msg, failure).build());
165 private <T> ListenableFuture<T> ask(ActorRef actor, Object message, Timeout timeout) {
166 final SettableFuture<T> returnFuture = SettableFuture.create();
168 @SuppressWarnings("unchecked")
169 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
170 askFuture.onComplete(new OnComplete<T>() {
172 public void onComplete(Throwable failure, T resp) {
173 if(failure != null) {
174 returnFuture.setException(failure);
176 returnFuture.set(resp);
179 }, configDataStore.getActorContext().getClientDispatcher());
184 private static RpcResultBuilder<Void> newFailedRpcResultBuilder(String message) {
185 return newFailedRpcResultBuilder(message, null);
188 private static RpcResultBuilder<Void> newFailedRpcResultBuilder(String message, Throwable cause) {
189 return RpcResultBuilder.<Void>failed().withError(ErrorType.RPC, message, cause);
192 private static RpcResult<Void> newSuccessfulResult() {
193 return RpcResultBuilder.<Void>success().build();