Adjust for Binding RPC codegen changes
[controller.git] / opendaylight / md-sal / sal-cluster-admin-impl / 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.MoreExecutors;
22 import com.google.common.util.concurrent.SettableFuture;
23 import java.io.FileOutputStream;
24 import java.io.IOException;
25 import java.util.AbstractMap.SimpleEntry;
26 import java.util.ArrayList;
27 import java.util.HashMap;
28 import java.util.List;
29 import java.util.Map;
30 import java.util.Map.Entry;
31 import java.util.Set;
32 import java.util.concurrent.TimeUnit;
33 import org.apache.commons.lang3.SerializationUtils;
34 import org.opendaylight.controller.cluster.access.concepts.MemberName;
35 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
36 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
37 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
38 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
39 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
40 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
41 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
42 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
43 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
44 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
45 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
46 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
47 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
48 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
49 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
50 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
51 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutput;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutputBuilder;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutput;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutputBuilder;
89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
95 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
96 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
97 import org.opendaylight.yangtools.yang.common.RpcResult;
98 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
99 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
100 import org.slf4j.Logger;
101 import org.slf4j.LoggerFactory;
102
103 /**
104  * Implements the yang RPCs defined in the generated ClusterAdminService interface.
105  *
106  * @author Thomas Pantelis
107  */
108 public class ClusterAdminRpcService implements ClusterAdminService {
109     private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
110
111     private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
112
113     private final DistributedDataStoreInterface configDataStore;
114     private final DistributedDataStoreInterface operDataStore;
115     private final BindingNormalizedNodeSerializer serializer;
116     private final Timeout makeLeaderLocalTimeout;
117
118     public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
119             final DistributedDataStoreInterface operDataStore,
120             final BindingNormalizedNodeSerializer serializer) {
121         this.configDataStore = configDataStore;
122         this.operDataStore = operDataStore;
123         this.serializer = serializer;
124
125         this.makeLeaderLocalTimeout =
126                 new Timeout(configDataStore.getActorContext().getDatastoreContext()
127                         .getShardLeaderElectionTimeout().duration().$times(2));
128     }
129
130     @Override
131     public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
132         final String shardName = input.getShardName();
133         if (Strings.isNullOrEmpty(shardName)) {
134             return newFailedRpcResultFuture("A valid shard name must be specified");
135         }
136
137         DataStoreType dataStoreType = input.getDataStoreType();
138         if (dataStoreType == null) {
139             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
140         }
141
142         LOG.info("Adding replica for shard {}", shardName);
143
144         final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
145         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
146         Futures.addCallback(future, new FutureCallback<Success>() {
147             @Override
148             public void onSuccess(final Success success) {
149                 LOG.info("Successfully added replica for shard {}", shardName);
150                 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
151             }
152
153             @Override
154             public void onFailure(final Throwable failure) {
155                 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
156                         returnFuture, failure);
157             }
158         }, MoreExecutors.directExecutor());
159
160         return returnFuture;
161     }
162
163     @Override
164     public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
165             final RemoveShardReplicaInput input) {
166         final String shardName = input.getShardName();
167         if (Strings.isNullOrEmpty(shardName)) {
168             return newFailedRpcResultFuture("A valid shard name must be specified");
169         }
170
171         DataStoreType dataStoreType = input.getDataStoreType();
172         if (dataStoreType == null) {
173             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
174         }
175
176         final String memberName = input.getMemberName();
177         if (Strings.isNullOrEmpty(memberName)) {
178             return newFailedRpcResultFuture("A valid member name must be specified");
179         }
180
181         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
182
183         final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
184         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
185                 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
186         Futures.addCallback(future, new FutureCallback<Success>() {
187             @Override
188             public void onSuccess(final Success success) {
189                 LOG.info("Successfully removed replica for shard {}", shardName);
190                 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
191             }
192
193             @Override
194             public void onFailure(final Throwable failure) {
195                 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
196                         returnFuture, failure);
197             }
198         }, MoreExecutors.directExecutor());
199
200         return returnFuture;
201     }
202
203     @Override
204     public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
205         final String shardName = input.getShardName();
206         if (Strings.isNullOrEmpty(shardName)) {
207             return newFailedRpcResultFuture("A valid shard name must be specified");
208         }
209
210         DataStoreType dataStoreType = input.getDataStoreType();
211         if (dataStoreType == null) {
212             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
213         }
214
215         ActorContext actorContext = dataStoreType == DataStoreType.Config
216                 ? configDataStore.getActorContext()
217                 : operDataStore.getActorContext();
218
219         LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
220                 actorContext.getCurrentMemberName().getName(), shardName, dataStoreType);
221
222         final scala.concurrent.Future<ActorRef> localShardReply =
223                 actorContext.findLocalShardAsync(shardName);
224
225         final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
226         localShardReply.onComplete(new OnComplete<ActorRef>() {
227             @Override
228             public void onComplete(final Throwable failure, final ActorRef actorRef) throws Throwable {
229                 if (failure != null) {
230                     LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
231                                     + " local shard.", shardName, failure);
232                     makeLeaderLocalAsk.failure(failure);
233                 } else {
234                     makeLeaderLocalAsk
235                             .completeWith(actorContext
236                                     .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
237                 }
238             }
239         }, actorContext.getClientDispatcher());
240
241         final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
242         makeLeaderLocalAsk.future().onComplete(new OnComplete<Object>() {
243             @Override
244             public void onComplete(final Throwable failure, final Object success) throws Throwable {
245                 if (failure != null) {
246                     LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
247                     future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
248                             "leadership transfer failed", failure).build());
249                     return;
250                 }
251
252                 LOG.debug("Leadership transfer complete");
253                 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
254             }
255         }, actorContext.getClientDispatcher());
256
257         return future;
258     }
259
260     @Override
261     public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
262             final AddPrefixShardReplicaInput input) {
263
264         final InstanceIdentifier<?> identifier = input.getShardPrefix();
265         if (identifier == null) {
266             return newFailedRpcResultFuture("A valid shard identifier must be specified");
267         }
268
269         final DataStoreType dataStoreType = input.getDataStoreType();
270         if (dataStoreType == null) {
271             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
272         }
273
274         LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
275
276         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
277         final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
278         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
279         Futures.addCallback(future, new FutureCallback<Success>() {
280             @Override
281             public void onSuccess(final Success success) {
282                 LOG.info("Successfully added replica for shard {}", prefix);
283                 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
284             }
285
286             @Override
287             public void onFailure(final Throwable failure) {
288                 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
289                         returnFuture, failure);
290             }
291         }, MoreExecutors.directExecutor());
292
293         return returnFuture;
294     }
295
296     @Override
297     public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
298             final RemovePrefixShardReplicaInput input) {
299
300         final InstanceIdentifier<?> identifier = input.getShardPrefix();
301         if (identifier == null) {
302             return newFailedRpcResultFuture("A valid shard identifier must be specified");
303         }
304
305         final DataStoreType dataStoreType = input.getDataStoreType();
306         if (dataStoreType == null) {
307             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
308         }
309
310         final String memberName = input.getMemberName();
311         if (Strings.isNullOrEmpty(memberName)) {
312             return newFailedRpcResultFuture("A valid member name must be specified");
313         }
314
315         LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
316                 identifier, memberName, dataStoreType);
317         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
318
319         final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
320         final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
321                 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
322         Futures.addCallback(future, new FutureCallback<Success>() {
323             @Override
324             public void onSuccess(final Success success) {
325                 LOG.info("Successfully removed replica for shard {}", prefix);
326                 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
327             }
328
329             @Override
330             public void onFailure(final Throwable failure) {
331                 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
332                         returnFuture, failure);
333             }
334         }, MoreExecutors.directExecutor());
335
336         return returnFuture;
337     }
338
339     @Override
340     public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
341             final AddReplicasForAllShardsInput input) {
342         LOG.info("Adding replicas for all shards");
343
344         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
345         Function<String, Object> messageSupplier = AddShardReplica::new;
346
347         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
348         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
349
350         return waitForShardResults(shardResultData, shardResults ->
351                 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
352                 "Failed to add replica");
353     }
354
355
356     @Override
357     public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
358             final RemoveAllShardReplicasInput input) {
359         LOG.info("Removing replicas for all shards");
360
361         final String memberName = input.getMemberName();
362         if (Strings.isNullOrEmpty(memberName)) {
363             return newFailedRpcResultFuture("A valid member name must be specified");
364         }
365
366         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
367         Function<String, Object> messageSupplier = shardName ->
368                 new RemoveShardReplica(shardName, MemberName.forName(memberName));
369
370         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
371         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
372
373         return waitForShardResults(shardResultData, shardResults ->
374                 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
375         "       Failed to remove replica");
376     }
377
378     @Override
379     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
380             final ChangeMemberVotingStatesForShardInput input) {
381         final String shardName = input.getShardName();
382         if (Strings.isNullOrEmpty(shardName)) {
383             return newFailedRpcResultFuture("A valid shard name must be specified");
384         }
385
386         DataStoreType dataStoreType = input.getDataStoreType();
387         if (dataStoreType == null) {
388             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
389         }
390
391         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
392         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
393             return newFailedRpcResultFuture("No member voting state input was specified");
394         }
395
396         ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
397                 memberVotingStates);
398
399         LOG.info("Change member voting states for shard {}: {}", shardName,
400                 changeVotingStatus.getMeberVotingStatusMap());
401
402         final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
403         ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
404         Futures.addCallback(future, new FutureCallback<Success>() {
405             @Override
406             public void onSuccess(final Success success) {
407                 LOG.info("Successfully changed member voting states for shard {}", shardName);
408                 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
409             }
410
411             @Override
412             public void onFailure(final Throwable failure) {
413                 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
414                         returnFuture, failure);
415             }
416         }, MoreExecutors.directExecutor());
417
418         return returnFuture;
419     }
420
421     @Override
422     public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
423             final ChangeMemberVotingStatesForAllShardsInput input) {
424         List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
425         if (memberVotingStates == null || memberVotingStates.isEmpty()) {
426             return newFailedRpcResultFuture("No member voting state input was specified");
427         }
428
429         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
430         Function<String, Object> messageSupplier = shardName ->
431                 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
432
433         LOG.info("Change member voting states for all shards");
434
435         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
436         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
437
438         return waitForShardResults(shardResultData, shardResults ->
439                 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
440                 "Failed to change member voting states");
441     }
442
443     @Override
444     public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
445             final FlipMemberVotingStatesForAllShardsInput input) {
446         final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
447         Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
448
449         LOG.info("Flip member voting states for all shards");
450
451         sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
452         sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
453
454         return waitForShardResults(shardResultData, shardResults ->
455                 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
456                 "Failed to change member voting states");
457     }
458
459     @Override
460     public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
461         final String shardName = input.getShardName();
462         if (Strings.isNullOrEmpty(shardName)) {
463             return newFailedRpcResultFuture("A valid shard name must be specified");
464         }
465
466         DataStoreType dataStoreType = input.getDataStoreType();
467         if (dataStoreType == null) {
468             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
469         }
470
471         LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
472
473         final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
474         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
475                 new GetShardRole(shardName));
476         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
477             @Override
478             public void onSuccess(final GetShardRoleReply reply) {
479                 if (reply == null) {
480                     returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
481                             "No Shard role present. Please retry..").build());
482                     return;
483                 }
484                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
485                 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
486                 if (reply.getRole() != null) {
487                     builder.setRole(reply.getRole());
488                 }
489                 returnFuture.set(newSuccessfulResult(builder.build()));
490             }
491
492             @Override
493             public void onFailure(final Throwable failure) {
494                 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
495                         "Failed to get shard role.", failure).build());
496             }
497         }, MoreExecutors.directExecutor());
498
499         return returnFuture;
500     }
501
502     @Override
503     public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
504             final GetPrefixShardRoleInput input) {
505         final InstanceIdentifier<?> identifier = input.getShardPrefix();
506         if (identifier == null) {
507             return newFailedRpcResultFuture("A valid shard identifier must be specified");
508         }
509
510         final DataStoreType dataStoreType = input.getDataStoreType();
511         if (dataStoreType == null) {
512             return newFailedRpcResultFuture("A valid DataStoreType must be specified");
513         }
514
515         LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
516
517         final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
518         final String shardName = ClusterUtils.getCleanShardName(prefix);
519         final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
520         ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
521                 new GetShardRole(shardName));
522         Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
523             @Override
524             public void onSuccess(final GetShardRoleReply reply) {
525                 if (reply == null) {
526                     returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
527                             "No Shard role present. Please retry..").build());
528                     return;
529                 }
530
531                 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
532                 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
533                 if (reply.getRole() != null) {
534                     builder.setRole(reply.getRole());
535                 }
536                 returnFuture.set(newSuccessfulResult(builder.build()));
537             }
538
539             @Override
540             public void onFailure(final Throwable failure) {
541                 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
542                         "Failed to get shard role.", failure).build());
543             }
544         }, MoreExecutors.directExecutor());
545
546         return returnFuture;
547     }
548
549     @Override
550     public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
551         LOG.debug("backupDatastore: {}", input);
552
553         if (Strings.isNullOrEmpty(input.getFilePath())) {
554             return newFailedRpcResultFuture("A valid file path must be specified");
555         }
556
557         final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
558         ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
559         Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
560             @Override
561             public void onSuccess(final List<DatastoreSnapshot> snapshots) {
562                 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
563             }
564
565             @Override
566             public void onFailure(final Throwable failure) {
567                 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
568             }
569         }, MoreExecutors.directExecutor());
570
571         return returnFuture;
572     }
573
574     private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
575             final List<MemberVotingState> memberVotingStatus) {
576         Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
577         for (MemberVotingState memberStatus: memberVotingStatus) {
578             serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
579         }
580         return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
581     }
582
583     private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
584             final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
585             final Function<List<ShardResult>, T> resultDataSupplier,
586             final String failureLogMsgPrefix) {
587         final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
588         final List<ShardResult> shardResults = new ArrayList<>();
589         for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
590             Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
591                 @Override
592                 public void onSuccess(final Success result) {
593                     synchronized (shardResults) {
594                         ShardResultBuilder shardResult = entry.getValue();
595                         LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
596                                 shardResult.getDataStoreType());
597                         shardResults.add(shardResult.setSucceeded(true).build());
598                         checkIfComplete();
599                     }
600                 }
601
602                 @Override
603                 public void onFailure(final Throwable failure) {
604                     synchronized (shardResults) {
605                         ShardResultBuilder shardResult = entry.getValue();
606                         LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
607                                 shardResult.getDataStoreType(), failure);
608                         shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
609                                 Throwables.getRootCause(failure).getMessage()).build());
610                         checkIfComplete();
611                     }
612                 }
613
614                 void checkIfComplete() {
615                     LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
616                     if (shardResults.size() == shardResultData.size()) {
617                         returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
618                     }
619                 }
620             }, MoreExecutors.directExecutor());
621         }
622         return returnFuture;
623     }
624
625     private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
626             final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
627             final Function<String, Object> messageSupplier) {
628         ActorContext actorContext = dataStoreType == DataStoreType.Config ? configDataStore.getActorContext()
629                 : operDataStore.getActorContext();
630         Set<String> allShardNames = actorContext.getConfiguration().getAllShardNames();
631
632         LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorContext.getDataStoreName());
633
634         for (String shardName: allShardNames) {
635             ListenableFuture<T> future = this.ask(actorContext.getShardManager(), messageSupplier.apply(shardName),
636                                                   SHARD_MGR_TIMEOUT);
637             shardResultData.add(new SimpleEntry<>(future,
638                     new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
639         }
640     }
641
642     private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
643         Timeout timeout = SHARD_MGR_TIMEOUT;
644         ListenableFuture<T> configFuture = ask(configDataStore.getActorContext().getShardManager(), message, timeout);
645         ListenableFuture<T> operFuture = ask(operDataStore.getActorContext().getShardManager(), message, timeout);
646
647         return Futures.allAsList(configFuture, operFuture);
648     }
649
650     private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
651         ActorRef shardManager = dataStoreType == DataStoreType.Config
652                 ? configDataStore.getActorContext().getShardManager()
653                         : operDataStore.getActorContext().getShardManager();
654         return ask(shardManager, message, SHARD_MGR_TIMEOUT);
655     }
656
657     @SuppressWarnings("checkstyle:IllegalCatch")
658     private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
659             final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
660         try (FileOutputStream fos = new FileOutputStream(fileName)) {
661             SerializationUtils.serialize(snapshots, fos);
662
663             returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
664             LOG.info("Successfully backed up datastore to file {}", fileName);
665         } catch (IOException | RuntimeException e) {
666             onDatastoreBackupFailure(fileName, returnFuture, e);
667         }
668     }
669
670     private static <T> void onDatastoreBackupFailure(final String fileName,
671             final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
672         onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
673     }
674
675     private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
676             final Throwable failure) {
677         LOG.error(msg, failure);
678         returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
679                 failure.getMessage())).build());
680     }
681
682     private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
683         final SettableFuture<T> returnFuture = SettableFuture.create();
684
685         @SuppressWarnings("unchecked")
686         scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
687         askFuture.onComplete(new OnComplete<T>() {
688             @Override
689             public void onComplete(final Throwable failure, final T resp) {
690                 if (failure != null) {
691                     returnFuture.setException(failure);
692                 } else {
693                     returnFuture.set(resp);
694                 }
695             }
696         }, configDataStore.getActorContext().getClientDispatcher());
697
698         return returnFuture;
699     }
700
701     private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
702         return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
703     }
704
705     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
706         return newFailedRpcResultBuilder(message, null);
707     }
708
709     private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
710         return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
711     }
712
713     private static <T> RpcResult<T> newSuccessfulResult(final T data) {
714         return RpcResultBuilder.success(data).build();
715     }
716 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.