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

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