2 * Copyright (c) 2015 Brocade Communications Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.datastore.admin;
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.collect.Maps;
20 import com.google.common.util.concurrent.FutureCallback;
21 import com.google.common.util.concurrent.Futures;
22 import com.google.common.util.concurrent.ListenableFuture;
23 import com.google.common.util.concurrent.MoreExecutors;
24 import com.google.common.util.concurrent.SettableFuture;
25 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
26 import java.io.FileOutputStream;
27 import java.io.IOException;
28 import java.util.AbstractMap.SimpleEntry;
29 import java.util.ArrayList;
30 import java.util.HashMap;
31 import java.util.List;
33 import java.util.Map.Entry;
35 import java.util.concurrent.ExecutionException;
36 import java.util.concurrent.TimeUnit;
37 import java.util.function.Function;
38 import java.util.stream.Collectors;
39 import org.apache.commons.lang3.SerializationUtils;
40 import org.eclipse.jdt.annotation.NonNull;
41 import org.opendaylight.controller.cluster.access.concepts.MemberName;
42 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
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.RemoveShardReplica;
53 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
54 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
55 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
56 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
57 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
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.GetKnownClientsForAllShardsInput;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutput;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutputBuilder;
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.RemoveShardReplicaInput;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
96 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;
97 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.KnownClients;
98 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;
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.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultKey;
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.common.Uint32;
110 import org.slf4j.Logger;
111 import org.slf4j.LoggerFactory;
112 import scala.concurrent.Future;
115 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
117 * @author Thomas Pantelis
119 public class ClusterAdminRpcService implements ClusterAdminService {
120 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
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())
129 private final DistributedDataStoreInterface configDataStore;
130 private final DistributedDataStoreInterface operDataStore;
131 private final BindingNormalizedNodeSerializer serializer;
132 private final Timeout makeLeaderLocalTimeout;
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;
141 this.makeLeaderLocalTimeout =
142 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
143 .getShardLeaderElectionTimeout().duration().$times(2));
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");
153 DataStoreType dataStoreType = input.getDataStoreType();
154 if (dataStoreType == null) {
155 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
158 LOG.info("Adding replica for shard {}", shardName);
160 final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
161 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
162 Futures.addCallback(future, new FutureCallback<Success>() {
164 public void onSuccess(final Success success) {
165 LOG.info("Successfully added replica for shard {}", shardName);
166 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
170 public void onFailure(final Throwable failure) {
171 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
172 returnFuture, failure);
174 }, MoreExecutors.directExecutor());
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");
187 DataStoreType dataStoreType = input.getDataStoreType();
188 if (dataStoreType == null) {
189 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
192 final String memberName = input.getMemberName();
193 if (Strings.isNullOrEmpty(memberName)) {
194 return newFailedRpcResultFuture("A valid member name must be specified");
197 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
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>() {
204 public void onSuccess(final Success success) {
205 LOG.info("Successfully removed replica for shard {}", shardName);
206 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
210 public void onFailure(final Throwable failure) {
211 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
212 returnFuture, failure);
214 }, MoreExecutors.directExecutor());
220 public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
221 final ActorUtils utils;
222 switch (input.getDataStoreType()) {
224 utils = configDataStore.getActorUtils();
227 utils = operDataStore.getActorUtils();
230 return newFailedRpcResultFuture("Unhandled datastore in " + input);
233 final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
234 utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
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);
243 // Data tree implies local leak
244 if (success.getLocalShardDataTree().isPresent()) {
245 ret.set(LOCAL_SHARD_RESULT);
249 final ActorSelection actorPath = success.getPrimaryShardActor();
250 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
251 .setMemberNode(new LeaderActorRefBuilder()
252 .setLeaderActorRef(actorPath.toSerializationFormat())
256 }, utils.getClientDispatcher());
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");
268 DataStoreType dataStoreType = input.getDataStoreType();
269 if (dataStoreType == null) {
270 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
273 ActorUtils actorUtils = dataStoreType == DataStoreType.Config
274 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
276 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
277 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
279 final Future<ActorRef> localShardReply = actorUtils.findLocalShardAsync(shardName);
281 final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
282 localShardReply.onComplete(new OnComplete<ActorRef>() {
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);
291 .completeWith(actorUtils
292 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
295 }, actorUtils.getClientDispatcher());
297 final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
298 makeLeaderLocalAsk.future().onComplete(new OnComplete<>() {
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());
308 LOG.debug("Leadership transfer complete");
309 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
311 }, actorUtils.getClientDispatcher());
317 public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
318 final AddReplicasForAllShardsInput input) {
319 LOG.info("Adding replicas for all shards");
321 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
323 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
324 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
326 return waitForShardResults(shardResultData, shardResults ->
327 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
328 "Failed to add replica");
333 public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
334 final RemoveAllShardReplicasInput input) {
335 LOG.info("Removing replicas for all shards");
337 final String memberName = input.getMemberName();
338 if (Strings.isNullOrEmpty(memberName)) {
339 return newFailedRpcResultFuture("A valid member name must be specified");
342 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
343 Function<String, Object> messageSupplier = shardName ->
344 new RemoveShardReplica(shardName, MemberName.forName(memberName));
346 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
347 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
349 return waitForShardResults(shardResultData, shardResults ->
350 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
351 " Failed to remove replica");
355 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
356 final ChangeMemberVotingStatesForShardInput input) {
357 final String shardName = input.getShardName();
358 if (Strings.isNullOrEmpty(shardName)) {
359 return newFailedRpcResultFuture("A valid shard name must be specified");
362 DataStoreType dataStoreType = input.getDataStoreType();
363 if (dataStoreType == null) {
364 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
367 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
368 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
369 return newFailedRpcResultFuture("No member voting state input was specified");
372 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
375 LOG.info("Change member voting states for shard {}: {}", shardName,
376 changeVotingStatus.getMeberVotingStatusMap());
378 final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
379 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
380 Futures.addCallback(future, new FutureCallback<Success>() {
382 public void onSuccess(final Success success) {
383 LOG.info("Successfully changed member voting states for shard {}", shardName);
384 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
388 public void onFailure(final Throwable failure) {
389 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
390 returnFuture, failure);
392 }, MoreExecutors.directExecutor());
398 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
399 final ChangeMemberVotingStatesForAllShardsInput input) {
400 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
401 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
402 return newFailedRpcResultFuture("No member voting state input was specified");
405 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
406 Function<String, Object> messageSupplier = shardName ->
407 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
409 LOG.info("Change member voting states for all shards");
411 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
412 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
414 return waitForShardResults(shardResultData, shardResults ->
415 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
416 "Failed to change member voting states");
420 public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
421 final FlipMemberVotingStatesForAllShardsInput input) {
422 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
423 Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
425 LOG.info("Flip member voting states for all shards");
427 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
428 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
430 return waitForShardResults(shardResultData, shardResults ->
431 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
432 "Failed to change member voting states");
436 public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
437 final String shardName = input.getShardName();
438 if (Strings.isNullOrEmpty(shardName)) {
439 return newFailedRpcResultFuture("A valid shard name must be specified");
442 DataStoreType dataStoreType = input.getDataStoreType();
443 if (dataStoreType == null) {
444 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
447 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
449 final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
450 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
451 new GetShardRole(shardName));
452 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
454 public void onSuccess(final GetShardRoleReply reply) {
456 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
457 "No Shard role present. Please retry..").build());
460 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
461 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
462 if (reply.getRole() != null) {
463 builder.setRole(reply.getRole());
465 returnFuture.set(newSuccessfulResult(builder.build()));
469 public void onFailure(final Throwable failure) {
470 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
471 "Failed to get shard role.", failure).build());
473 }, MoreExecutors.directExecutor());
479 public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
480 LOG.debug("backupDatastore: {}", input);
482 if (Strings.isNullOrEmpty(input.getFilePath())) {
483 return newFailedRpcResultFuture("A valid file path must be specified");
486 final Uint32 timeout = input.getTimeout();
487 final Timeout opTimeout = timeout != null ? Timeout.apply(timeout.longValue(), TimeUnit.SECONDS)
490 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
491 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(new GetSnapshot(opTimeout));
492 Futures.addCallback(future, new FutureCallback<>() {
494 public void onSuccess(final List<DatastoreSnapshot> snapshots) {
495 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
499 public void onFailure(final Throwable failure) {
500 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
502 }, MoreExecutors.directExecutor());
509 public ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
510 final GetKnownClientsForAllShardsInput input) {
511 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies =
512 getAllShardLeadersClients();
513 return Futures.whenAllComplete(allShardReplies.values()).call(() -> processReplies(allShardReplies),
514 MoreExecutors.directExecutor());
517 private static RpcResult<GetKnownClientsForAllShardsOutput> processReplies(
518 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies) {
519 final Map<ShardResultKey, ShardResult> result = Maps.newHashMapWithExpectedSize(allShardReplies.size());
520 for (Entry<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> entry : allShardReplies.entrySet()) {
521 final ListenableFuture<GetKnownClientsReply> future = entry.getValue();
522 final ShardResultBuilder builder = new ShardResultBuilder()
523 .setDataStoreType(entry.getKey().getDataStoreType())
524 .setShardName(entry.getKey().getShardName());
526 final GetKnownClientsReply reply;
528 reply = Futures.getDone(future);
529 } catch (ExecutionException e) {
530 LOG.debug("Shard {} failed to answer", entry.getKey(), e);
531 final ShardResult sr = builder
532 .setSucceeded(Boolean.FALSE)
533 .setErrorMessage(e.getCause().getMessage())
535 result.put(sr.key(), sr);
539 final ShardResult sr = builder
540 .setSucceeded(Boolean.TRUE)
541 .addAugmentation(new ShardResult1Builder()
542 .setKnownClients(reply.getClients().stream()
543 .map(client -> new KnownClientsBuilder()
544 .setMember(client.getFrontendId().getMemberName().toYang())
545 .setType(client.getFrontendId().getClientType().toYang())
546 .setGeneration(client.getYangGeneration())
548 .collect(Collectors.toMap(KnownClients::key, Function.identity())))
552 result.put(sr.key(), sr);
555 return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
559 private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
560 final List<MemberVotingState> memberVotingStatus) {
561 Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
562 for (MemberVotingState memberStatus: memberVotingStatus) {
563 serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.getVoting());
565 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
568 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
569 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
570 final Function<Map<ShardResultKey, ShardResult>, T> resultDataSupplier,
571 final String failureLogMsgPrefix) {
572 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
573 final Map<ShardResultKey, ShardResult> shardResults = new HashMap<>();
574 for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
575 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
577 public void onSuccess(final Success result) {
578 synchronized (shardResults) {
579 final ShardResultBuilder builder = entry.getValue();
580 LOG.debug("onSuccess for shard {}, type {}", builder.getShardName(),
581 builder.getDataStoreType());
582 final ShardResult sr = builder.setSucceeded(Boolean.TRUE).build();
583 shardResults.put(sr.key(), sr);
589 public void onFailure(final Throwable failure) {
590 synchronized (shardResults) {
591 ShardResultBuilder builder = entry.getValue();
592 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, builder.getShardName(),
593 builder.getDataStoreType(), failure);
594 final ShardResult sr = builder
595 .setSucceeded(Boolean.FALSE)
596 .setErrorMessage(Throwables.getRootCause(failure).getMessage())
598 shardResults.put(sr.key(), sr);
603 void checkIfComplete() {
604 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
605 if (shardResults.size() == shardResultData.size()) {
606 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
609 }, MoreExecutors.directExecutor());
614 private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
615 final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
616 final Function<String, Object> messageSupplier) {
617 ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
618 : operDataStore.getActorUtils();
619 Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
621 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
623 for (String shardName: allShardNames) {
624 ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
626 shardResultData.add(new SimpleEntry<>(future,
627 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
631 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
632 Timeout timeout = SHARD_MGR_TIMEOUT;
633 ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
634 ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
636 return Futures.allAsList(configFuture, operFuture);
639 private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
640 ActorRef shardManager = dataStoreType == DataStoreType.Config
641 ? configDataStore.getActorUtils().getShardManager()
642 : operDataStore.getActorUtils().getShardManager();
643 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
646 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
647 justification = "https://github.com/spotbugs/spotbugs/issues/811")
648 @SuppressWarnings("checkstyle:IllegalCatch")
649 private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
650 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
651 try (FileOutputStream fos = new FileOutputStream(fileName)) {
652 SerializationUtils.serialize(snapshots, fos);
654 returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
655 LOG.info("Successfully backed up datastore to file {}", fileName);
656 } catch (IOException | RuntimeException e) {
657 onDatastoreBackupFailure(fileName, returnFuture, e);
661 private static <T> void onDatastoreBackupFailure(final String fileName,
662 final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
663 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
666 @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
667 private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
668 final Throwable failure) {
669 LOG.error("{}", msg, failure);
670 returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
671 failure.getMessage())).build());
674 private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
675 final SettableFuture<T> returnFuture = SettableFuture.create();
677 @SuppressWarnings("unchecked")
678 Future<T> askFuture = (Future<T>) Patterns.ask(actor, message, timeout);
679 askFuture.onComplete(new OnComplete<T>() {
681 public void onComplete(final Throwable failure, final T resp) {
682 if (failure != null) {
683 returnFuture.setException(failure);
685 returnFuture.set(resp);
688 }, configDataStore.getActorUtils().getClientDispatcher());
693 private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
694 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
695 ImmutableMap.builder();
697 addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
698 addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
700 return builder.build();
703 private static void addAllShardsClients(
704 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder,
705 final DataStoreType type, final ActorUtils utils) {
706 for (String shardName : utils.getConfiguration().getAllShardNames()) {
707 final SettableFuture<GetKnownClientsReply> future = SettableFuture.create();
708 builder.put(new ShardIdentifier(type, shardName), future);
710 utils.findPrimaryShardAsync(shardName).flatMap(
711 info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
712 utils.getClientDispatcher()).onComplete(new OnComplete<>() {
714 public void onComplete(final Throwable failure, final Object success) {
715 if (failure == null) {
716 future.set((GetKnownClientsReply) success);
718 future.setException(failure);
721 }, utils.getClientDispatcher());
725 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
726 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
729 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
730 return newFailedRpcResultBuilder(message, null);
733 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
734 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
737 private static <T> RpcResult<T> newSuccessfulResult(final T data) {
738 return RpcResultBuilder.success(data).build();