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

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