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

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