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

©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.