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.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;
32 import java.util.Map.Entry;
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;
126 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
128 * @author Thomas Pantelis
130 public class ClusterAdminRpcService implements ClusterAdminService {
131 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
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())
140 private final DistributedDataStoreInterface configDataStore;
141 private final DistributedDataStoreInterface operDataStore;
142 private final BindingNormalizedNodeSerializer serializer;
143 private final Timeout makeLeaderLocalTimeout;
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;
152 this.makeLeaderLocalTimeout =
153 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
154 .getShardLeaderElectionTimeout().duration().$times(2));
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");
164 DataStoreType dataStoreType = input.getDataStoreType();
165 if (dataStoreType == null) {
166 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
169 LOG.info("Adding replica for shard {}", shardName);
171 final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
172 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
173 Futures.addCallback(future, new FutureCallback<Success>() {
175 public void onSuccess(final Success success) {
176 LOG.info("Successfully added replica for shard {}", shardName);
177 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
181 public void onFailure(final Throwable failure) {
182 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
183 returnFuture, failure);
185 }, MoreExecutors.directExecutor());
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");
198 DataStoreType dataStoreType = input.getDataStoreType();
199 if (dataStoreType == null) {
200 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
203 final String memberName = input.getMemberName();
204 if (Strings.isNullOrEmpty(memberName)) {
205 return newFailedRpcResultFuture("A valid member name must be specified");
208 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
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>() {
215 public void onSuccess(final Success success) {
216 LOG.info("Successfully removed replica for shard {}", shardName);
217 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
221 public void onFailure(final Throwable failure) {
222 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
223 returnFuture, failure);
225 }, MoreExecutors.directExecutor());
231 public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
232 final ActorUtils utils;
233 switch (input.getDataStoreType()) {
235 utils = configDataStore.getActorUtils();
238 utils = operDataStore.getActorUtils();
241 return newFailedRpcResultFuture("Unhandled datastore in " + input);
244 final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
245 utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
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);
254 // Data tree implies local leak
255 if (success.getLocalShardDataTree().isPresent()) {
256 ret.set(LOCAL_SHARD_RESULT);
260 final ActorSelection actorPath = success.getPrimaryShardActor();
261 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
262 .setMemberNode(new LeaderActorRefBuilder()
263 .setLeaderActorRef(actorPath.toSerializationFormat())
267 }, utils.getClientDispatcher());
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");
279 DataStoreType dataStoreType = input.getDataStoreType();
280 if (dataStoreType == null) {
281 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
284 ActorUtils actorUtils = dataStoreType == DataStoreType.Config
285 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
287 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
288 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
290 final Future<ActorRef> localShardReply = actorUtils.findLocalShardAsync(shardName);
292 final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
293 localShardReply.onComplete(new OnComplete<ActorRef>() {
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);
302 .completeWith(actorUtils
303 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
306 }, actorUtils.getClientDispatcher());
308 final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
309 makeLeaderLocalAsk.future().onComplete(new OnComplete<>() {
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());
319 LOG.debug("Leadership transfer complete");
320 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
322 }, actorUtils.getClientDispatcher());
328 public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
329 final AddPrefixShardReplicaInput input) {
331 final InstanceIdentifier<?> identifier = input.getShardPrefix();
332 if (identifier == null) {
333 return newFailedRpcResultFuture("A valid shard identifier must be specified");
336 final DataStoreType dataStoreType = input.getDataStoreType();
337 if (dataStoreType == null) {
338 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
341 LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
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>() {
348 public void onSuccess(final Success success) {
349 LOG.info("Successfully added replica for shard {}", prefix);
350 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
354 public void onFailure(final Throwable failure) {
355 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
356 returnFuture, failure);
358 }, MoreExecutors.directExecutor());
364 public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
365 final RemovePrefixShardReplicaInput input) {
367 final InstanceIdentifier<?> identifier = input.getShardPrefix();
368 if (identifier == null) {
369 return newFailedRpcResultFuture("A valid shard identifier must be specified");
372 final DataStoreType dataStoreType = input.getDataStoreType();
373 if (dataStoreType == null) {
374 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
377 final String memberName = input.getMemberName();
378 if (Strings.isNullOrEmpty(memberName)) {
379 return newFailedRpcResultFuture("A valid member name must be specified");
382 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
383 identifier, memberName, dataStoreType);
384 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
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>() {
391 public void onSuccess(final Success success) {
392 LOG.info("Successfully removed replica for shard {}", prefix);
393 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
397 public void onFailure(final Throwable failure) {
398 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
399 returnFuture, failure);
401 }, MoreExecutors.directExecutor());
407 public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
408 final AddReplicasForAllShardsInput input) {
409 LOG.info("Adding replicas for all shards");
411 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
413 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
414 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
416 return waitForShardResults(shardResultData, shardResults ->
417 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
418 "Failed to add replica");
423 public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
424 final RemoveAllShardReplicasInput input) {
425 LOG.info("Removing replicas for all shards");
427 final String memberName = input.getMemberName();
428 if (Strings.isNullOrEmpty(memberName)) {
429 return newFailedRpcResultFuture("A valid member name must be specified");
432 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
433 Function<String, Object> messageSupplier = shardName ->
434 new RemoveShardReplica(shardName, MemberName.forName(memberName));
436 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
437 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
439 return waitForShardResults(shardResultData, shardResults ->
440 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
441 " Failed to remove replica");
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");
452 DataStoreType dataStoreType = input.getDataStoreType();
453 if (dataStoreType == null) {
454 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
457 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
458 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
459 return newFailedRpcResultFuture("No member voting state input was specified");
462 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
465 LOG.info("Change member voting states for shard {}: {}", shardName,
466 changeVotingStatus.getMeberVotingStatusMap());
468 final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
469 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
470 Futures.addCallback(future, new FutureCallback<Success>() {
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()));
478 public void onFailure(final Throwable failure) {
479 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
480 returnFuture, failure);
482 }, MoreExecutors.directExecutor());
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");
495 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
496 Function<String, Object> messageSupplier = shardName ->
497 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
499 LOG.info("Change member voting states for all shards");
501 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
502 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
504 return waitForShardResults(shardResultData, shardResults ->
505 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
506 "Failed to change member voting states");
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;
515 LOG.info("Flip member voting states for all shards");
517 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
518 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
520 return waitForShardResults(shardResultData, shardResults ->
521 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
522 "Failed to change member voting states");
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");
532 DataStoreType dataStoreType = input.getDataStoreType();
533 if (dataStoreType == null) {
534 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
537 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
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>() {
544 public void onSuccess(final GetShardRoleReply reply) {
546 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
547 "No Shard role present. Please retry..").build());
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());
555 returnFuture.set(newSuccessfulResult(builder.build()));
559 public void onFailure(final Throwable failure) {
560 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
561 "Failed to get shard role.", failure).build());
563 }, MoreExecutors.directExecutor());
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");
576 final DataStoreType dataStoreType = input.getDataStoreType();
577 if (dataStoreType == null) {
578 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
581 LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
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>() {
590 public void onSuccess(final GetShardRoleReply reply) {
592 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
593 "No Shard role present. Please retry..").build());
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());
602 returnFuture.set(newSuccessfulResult(builder.build()));
606 public void onFailure(final Throwable failure) {
607 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
608 "Failed to get shard role.", failure).build());
610 }, MoreExecutors.directExecutor());
616 public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
617 LOG.debug("backupDatastore: {}", input);
619 if (Strings.isNullOrEmpty(input.getFilePath())) {
620 return newFailedRpcResultFuture("A valid file path must be specified");
623 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
624 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
625 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
627 public void onSuccess(final List<DatastoreSnapshot> snapshots) {
628 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
632 public void onFailure(final Throwable failure) {
633 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
635 }, MoreExecutors.directExecutor());
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());
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());
659 final GetKnownClientsReply reply;
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());
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())
677 .collect(Collectors.toList()))
682 return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
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());
692 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
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>() {
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());
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());
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)));
732 }, MoreExecutors.directExecutor());
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();
744 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
746 for (String shardName: allShardNames) {
747 ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
749 shardResultData.add(new SimpleEntry<>(future,
750 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
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);
759 return Futures.allAsList(configFuture, operFuture);
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);
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);
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);
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);
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());
797 private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
798 final SettableFuture<T> returnFuture = SettableFuture.create();
800 @SuppressWarnings("unchecked")
801 Future<T> askFuture = (Future<T>) Patterns.ask(actor, message, timeout);
802 askFuture.onComplete(new OnComplete<T>() {
804 public void onComplete(final Throwable failure, final T resp) {
805 if (failure != null) {
806 returnFuture.setException(failure);
808 returnFuture.set(resp);
811 }, configDataStore.getActorUtils().getClientDispatcher());
816 private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
817 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
818 ImmutableMap.builder();
820 addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
821 addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
823 return builder.build();
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);
833 utils.findPrimaryShardAsync(shardName).flatMap(
834 info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
835 utils.getClientDispatcher()).onComplete(new OnComplete<>() {
837 public void onComplete(final Throwable failure, final Object success) {
838 if (failure == null) {
839 future.set((GetKnownClientsReply) success);
841 future.setException(failure);
844 }, utils.getClientDispatcher());
848 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
849 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
852 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
853 return newFailedRpcResultBuilder(message, null);
856 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
857 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
860 private static <T> RpcResult<T> newSuccessfulResult(final T data) {
861 return RpcResultBuilder.success(data).build();