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.controller.eos.akka.NativeEosService;
58 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ActivateEosDatacenterInput;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ActivateEosDatacenterOutput;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DeactivateEosDatacenterInput;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DeactivateEosDatacenterOutput;
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.GetShardRoleInput;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardInput;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutput;
91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutputBuilder;
92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
97 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
98 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
99 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
100 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
101 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;
102 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;
103 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;
104 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LeaderActorRefBuilder;
105 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LocalBuilder;
106 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
107 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
108 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
109 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultKey;
110 import org.opendaylight.yangtools.yang.common.Empty;
111 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
112 import org.opendaylight.yangtools.yang.common.RpcResult;
113 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
114 import org.opendaylight.yangtools.yang.common.Uint32;
115 import org.slf4j.Logger;
116 import org.slf4j.LoggerFactory;
117 import scala.concurrent.Future;
120 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
122 * @author Thomas Pantelis
124 public class ClusterAdminRpcService implements ClusterAdminService {
125 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
127 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
128 private static final @NonNull RpcResult<LocateShardOutput> LOCAL_SHARD_RESULT =
129 RpcResultBuilder.success(new LocateShardOutputBuilder()
130 .setMemberNode(new LocalBuilder().setLocal(Empty.getInstance()).build())
134 private final DistributedDataStoreInterface configDataStore;
135 private final DistributedDataStoreInterface operDataStore;
136 private final BindingNormalizedNodeSerializer serializer;
137 private final Timeout makeLeaderLocalTimeout;
138 private final NativeEosService nativeEosService;
140 public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
141 final DistributedDataStoreInterface operDataStore,
142 final BindingNormalizedNodeSerializer serializer,
143 final NativeEosService nativeEosService) {
144 this.configDataStore = configDataStore;
145 this.operDataStore = operDataStore;
146 this.serializer = serializer;
148 this.makeLeaderLocalTimeout =
149 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
150 .getShardLeaderElectionTimeout().duration().$times(2));
152 this.nativeEosService = nativeEosService;
156 public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
157 final String shardName = input.getShardName();
158 if (Strings.isNullOrEmpty(shardName)) {
159 return newFailedRpcResultFuture("A valid shard name must be specified");
162 DataStoreType dataStoreType = input.getDataStoreType();
163 if (dataStoreType == null) {
164 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
167 LOG.info("Adding replica for shard {}", shardName);
169 final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
170 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
171 Futures.addCallback(future, new FutureCallback<Success>() {
173 public void onSuccess(final Success success) {
174 LOG.info("Successfully added replica for shard {}", shardName);
175 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
179 public void onFailure(final Throwable failure) {
180 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
181 returnFuture, failure);
183 }, MoreExecutors.directExecutor());
189 public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
190 final RemoveShardReplicaInput input) {
191 final String shardName = input.getShardName();
192 if (Strings.isNullOrEmpty(shardName)) {
193 return newFailedRpcResultFuture("A valid shard name must be specified");
196 DataStoreType dataStoreType = input.getDataStoreType();
197 if (dataStoreType == null) {
198 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
201 final String memberName = input.getMemberName();
202 if (Strings.isNullOrEmpty(memberName)) {
203 return newFailedRpcResultFuture("A valid member name must be specified");
206 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
208 final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
209 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
210 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
211 Futures.addCallback(future, new FutureCallback<Success>() {
213 public void onSuccess(final Success success) {
214 LOG.info("Successfully removed replica for shard {}", shardName);
215 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
219 public void onFailure(final Throwable failure) {
220 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
221 returnFuture, failure);
223 }, MoreExecutors.directExecutor());
229 public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
230 final ActorUtils utils;
231 switch (input.getDataStoreType()) {
233 utils = configDataStore.getActorUtils();
236 utils = operDataStore.getActorUtils();
239 return newFailedRpcResultFuture("Unhandled datastore in " + input);
242 final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
243 utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
245 public void onComplete(final Throwable failure, final PrimaryShardInfo success) throws Throwable {
246 if (failure != null) {
247 LOG.debug("Failed to find shard for {}", input, failure);
248 ret.setException(failure);
252 // Data tree implies local leak
253 if (success.getLocalShardDataTree().isPresent()) {
254 ret.set(LOCAL_SHARD_RESULT);
258 final ActorSelection actorPath = success.getPrimaryShardActor();
259 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
260 .setMemberNode(new LeaderActorRefBuilder()
261 .setLeaderActorRef(actorPath.toSerializationFormat())
265 }, utils.getClientDispatcher());
271 public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
272 final String shardName = input.getShardName();
273 if (Strings.isNullOrEmpty(shardName)) {
274 return newFailedRpcResultFuture("A valid shard name must be specified");
277 DataStoreType dataStoreType = input.getDataStoreType();
278 if (dataStoreType == null) {
279 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
282 ActorUtils actorUtils = dataStoreType == DataStoreType.Config
283 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
285 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
286 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
288 final Future<ActorRef> localShardReply = actorUtils.findLocalShardAsync(shardName);
290 final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
291 localShardReply.onComplete(new OnComplete<ActorRef>() {
293 public void onComplete(final Throwable failure, final ActorRef actorRef) {
294 if (failure != null) {
295 LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
296 + " local shard.", shardName, dataStoreType, failure);
297 makeLeaderLocalAsk.failure(failure);
300 .completeWith(actorUtils
301 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
304 }, actorUtils.getClientDispatcher());
306 final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
307 makeLeaderLocalAsk.future().onComplete(new OnComplete<>() {
309 public void onComplete(final Throwable failure, final Object success) {
310 if (failure != null) {
311 LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
312 future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
313 "leadership transfer failed", failure).build());
317 LOG.debug("Leadership transfer complete");
318 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
320 }, actorUtils.getClientDispatcher());
326 public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
327 final AddReplicasForAllShardsInput input) {
328 LOG.info("Adding replicas for all shards");
330 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
332 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
333 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
335 return waitForShardResults(shardResultData, shardResults ->
336 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
337 "Failed to add replica");
342 public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
343 final RemoveAllShardReplicasInput input) {
344 LOG.info("Removing replicas for all shards");
346 final String memberName = input.getMemberName();
347 if (Strings.isNullOrEmpty(memberName)) {
348 return newFailedRpcResultFuture("A valid member name must be specified");
351 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
352 Function<String, Object> messageSupplier = shardName ->
353 new RemoveShardReplica(shardName, MemberName.forName(memberName));
355 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
356 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
358 return waitForShardResults(shardResultData, shardResults ->
359 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
360 " Failed to remove replica");
364 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
365 final ChangeMemberVotingStatesForShardInput input) {
366 final String shardName = input.getShardName();
367 if (Strings.isNullOrEmpty(shardName)) {
368 return newFailedRpcResultFuture("A valid shard name must be specified");
371 DataStoreType dataStoreType = input.getDataStoreType();
372 if (dataStoreType == null) {
373 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
376 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
377 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
378 return newFailedRpcResultFuture("No member voting state input was specified");
381 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
384 LOG.info("Change member voting states for shard {}: {}", shardName,
385 changeVotingStatus.getMeberVotingStatusMap());
387 final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
388 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
389 Futures.addCallback(future, new FutureCallback<Success>() {
391 public void onSuccess(final Success success) {
392 LOG.info("Successfully changed member voting states for shard {}", shardName);
393 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
397 public void onFailure(final Throwable failure) {
398 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
399 returnFuture, failure);
401 }, MoreExecutors.directExecutor());
407 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
408 final ChangeMemberVotingStatesForAllShardsInput input) {
409 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
410 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
411 return newFailedRpcResultFuture("No member voting state input was specified");
414 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
415 Function<String, Object> messageSupplier = shardName ->
416 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
418 LOG.info("Change member voting states for all shards");
420 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
421 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
423 return waitForShardResults(shardResultData, shardResults ->
424 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
425 "Failed to change member voting states");
429 public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
430 final FlipMemberVotingStatesForAllShardsInput input) {
431 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
432 Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
434 LOG.info("Flip member voting states for all shards");
436 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
437 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
439 return waitForShardResults(shardResultData, shardResults ->
440 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
441 "Failed to change member voting states");
445 public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
446 final String shardName = input.getShardName();
447 if (Strings.isNullOrEmpty(shardName)) {
448 return newFailedRpcResultFuture("A valid shard name must be specified");
451 DataStoreType dataStoreType = input.getDataStoreType();
452 if (dataStoreType == null) {
453 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
456 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
458 final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
459 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
460 new GetShardRole(shardName));
461 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
463 public void onSuccess(final GetShardRoleReply reply) {
465 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
466 "No Shard role present. Please retry..").build());
469 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
470 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
471 if (reply.getRole() != null) {
472 builder.setRole(reply.getRole());
474 returnFuture.set(newSuccessfulResult(builder.build()));
478 public void onFailure(final Throwable failure) {
479 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
480 "Failed to get shard role.", failure).build());
482 }, MoreExecutors.directExecutor());
488 public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
489 LOG.debug("backupDatastore: {}", input);
491 if (Strings.isNullOrEmpty(input.getFilePath())) {
492 return newFailedRpcResultFuture("A valid file path must be specified");
495 final Uint32 timeout = input.getTimeout();
496 final Timeout opTimeout = timeout != null ? Timeout.apply(timeout.longValue(), TimeUnit.SECONDS)
499 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
500 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(new GetSnapshot(opTimeout));
501 Futures.addCallback(future, new FutureCallback<>() {
503 public void onSuccess(final List<DatastoreSnapshot> snapshots) {
504 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
508 public void onFailure(final Throwable failure) {
509 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
511 }, MoreExecutors.directExecutor());
518 public ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
519 final GetKnownClientsForAllShardsInput input) {
520 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies =
521 getAllShardLeadersClients();
522 return Futures.whenAllComplete(allShardReplies.values()).call(() -> processReplies(allShardReplies),
523 MoreExecutors.directExecutor());
527 public ListenableFuture<RpcResult<ActivateEosDatacenterOutput>> activateEosDatacenter(
528 final ActivateEosDatacenterInput input) {
529 LOG.debug("Activating EOS Datacenter");
530 final SettableFuture<RpcResult<ActivateEosDatacenterOutput>> future = SettableFuture.create();
531 Futures.addCallback(nativeEosService.activateDataCenter(), new FutureCallback<>() {
533 public void onSuccess(final Void result) {
534 LOG.debug("Successfully activated datacenter.");
535 future.set(RpcResultBuilder.<ActivateEosDatacenterOutput>success().build());
539 public void onFailure(final Throwable failure) {
540 future.set(ClusterAdminRpcService.<ActivateEosDatacenterOutput>newFailedRpcResultBuilder(
541 "Failed to activate datacenter.", failure).build());
543 }, MoreExecutors.directExecutor());
549 public ListenableFuture<RpcResult<DeactivateEosDatacenterOutput>> deactivateEosDatacenter(
550 final DeactivateEosDatacenterInput input) {
551 LOG.debug("Deactivating EOS Datacenter");
552 final SettableFuture<RpcResult<DeactivateEosDatacenterOutput>> future = SettableFuture.create();
553 Futures.addCallback(nativeEosService.deactivateDataCenter(), new FutureCallback<>() {
555 public void onSuccess(final Void result) {
556 LOG.debug("Successfully deactivated datacenter.");
557 future.set(RpcResultBuilder.<DeactivateEosDatacenterOutput>success().build());
561 public void onFailure(final Throwable failure) {
562 future.set(ClusterAdminRpcService.<DeactivateEosDatacenterOutput>newFailedRpcResultBuilder(
563 "Failed to deactivate datacenter.", failure).build());
565 }, MoreExecutors.directExecutor());
570 private static RpcResult<GetKnownClientsForAllShardsOutput> processReplies(
571 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies) {
572 final Map<ShardResultKey, ShardResult> result = Maps.newHashMapWithExpectedSize(allShardReplies.size());
573 for (Entry<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> entry : allShardReplies.entrySet()) {
574 final ListenableFuture<GetKnownClientsReply> future = entry.getValue();
575 final ShardResultBuilder builder = new ShardResultBuilder()
576 .setDataStoreType(entry.getKey().getDataStoreType())
577 .setShardName(entry.getKey().getShardName());
579 final GetKnownClientsReply reply;
581 reply = Futures.getDone(future);
582 } catch (ExecutionException e) {
583 LOG.debug("Shard {} failed to answer", entry.getKey(), e);
584 final ShardResult sr = builder
585 .setSucceeded(Boolean.FALSE)
586 .setErrorMessage(e.getCause().getMessage())
588 result.put(sr.key(), sr);
592 final ShardResult sr = builder
593 .setSucceeded(Boolean.TRUE)
594 .addAugmentation(new ShardResult1Builder()
595 .setKnownClients(reply.getClients().stream()
596 .map(client -> new KnownClientsBuilder()
597 .setMember(client.getFrontendId().getMemberName().toYang())
598 .setType(client.getFrontendId().getClientType().toYang())
599 .setGeneration(client.getYangGeneration())
601 .collect(Collectors.toMap(KnownClients::key, Function.identity())))
605 result.put(sr.key(), sr);
608 return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
612 private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
613 final List<MemberVotingState> memberVotingStatus) {
614 Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
615 for (MemberVotingState memberStatus: memberVotingStatus) {
616 serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.getVoting());
618 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
621 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
622 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
623 final Function<Map<ShardResultKey, ShardResult>, T> resultDataSupplier,
624 final String failureLogMsgPrefix) {
625 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
626 final Map<ShardResultKey, ShardResult> shardResults = new HashMap<>();
627 for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
628 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
630 public void onSuccess(final Success result) {
631 synchronized (shardResults) {
632 final ShardResultBuilder builder = entry.getValue();
633 LOG.debug("onSuccess for shard {}, type {}", builder.getShardName(),
634 builder.getDataStoreType());
635 final ShardResult sr = builder.setSucceeded(Boolean.TRUE).build();
636 shardResults.put(sr.key(), sr);
642 public void onFailure(final Throwable failure) {
643 synchronized (shardResults) {
644 ShardResultBuilder builder = entry.getValue();
645 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, builder.getShardName(),
646 builder.getDataStoreType(), failure);
647 final ShardResult sr = builder
648 .setSucceeded(Boolean.FALSE)
649 .setErrorMessage(Throwables.getRootCause(failure).getMessage())
651 shardResults.put(sr.key(), sr);
656 void checkIfComplete() {
657 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
658 if (shardResults.size() == shardResultData.size()) {
659 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
662 }, MoreExecutors.directExecutor());
667 private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
668 final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
669 final Function<String, Object> messageSupplier) {
670 ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
671 : operDataStore.getActorUtils();
672 Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
674 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
676 for (String shardName: allShardNames) {
677 ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
679 shardResultData.add(new SimpleEntry<>(future,
680 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
684 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
685 Timeout timeout = SHARD_MGR_TIMEOUT;
686 ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
687 ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
689 return Futures.allAsList(configFuture, operFuture);
692 private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
693 ActorRef shardManager = dataStoreType == DataStoreType.Config
694 ? configDataStore.getActorUtils().getShardManager()
695 : operDataStore.getActorUtils().getShardManager();
696 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
699 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
700 justification = "https://github.com/spotbugs/spotbugs/issues/811")
701 @SuppressWarnings("checkstyle:IllegalCatch")
702 private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
703 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
704 try (FileOutputStream fos = new FileOutputStream(fileName)) {
705 SerializationUtils.serialize(snapshots, fos);
707 returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
708 LOG.info("Successfully backed up datastore to file {}", fileName);
709 } catch (IOException | RuntimeException e) {
710 onDatastoreBackupFailure(fileName, returnFuture, e);
714 private static <T> void onDatastoreBackupFailure(final String fileName,
715 final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
716 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
719 @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
720 private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
721 final Throwable failure) {
722 LOG.error("{}", msg, failure);
723 returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
724 failure.getMessage())).build());
727 private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
728 final SettableFuture<T> returnFuture = SettableFuture.create();
730 @SuppressWarnings("unchecked")
731 Future<T> askFuture = (Future<T>) Patterns.ask(actor, message, timeout);
732 askFuture.onComplete(new OnComplete<T>() {
734 public void onComplete(final Throwable failure, final T resp) {
735 if (failure != null) {
736 returnFuture.setException(failure);
738 returnFuture.set(resp);
741 }, configDataStore.getActorUtils().getClientDispatcher());
746 private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
747 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
748 ImmutableMap.builder();
750 addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
751 addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
753 return builder.build();
756 private static void addAllShardsClients(
757 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder,
758 final DataStoreType type, final ActorUtils utils) {
759 for (String shardName : utils.getConfiguration().getAllShardNames()) {
760 final SettableFuture<GetKnownClientsReply> future = SettableFuture.create();
761 builder.put(new ShardIdentifier(type, shardName), future);
763 utils.findPrimaryShardAsync(shardName).flatMap(
764 info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
765 utils.getClientDispatcher()).onComplete(new OnComplete<>() {
767 public void onComplete(final Throwable failure, final Object success) {
768 if (failure == null) {
769 future.set((GetKnownClientsReply) success);
771 future.setException(failure);
774 }, utils.getClientDispatcher());
778 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
779 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
782 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
783 return newFailedRpcResultBuilder(message, null);
786 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
787 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
790 private static <T> RpcResult<T> newSuccessfulResult(final T data) {
791 return RpcResultBuilder.success(data).build();