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

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