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.util.concurrent.FutureCallback;
19 import com.google.common.util.concurrent.Futures;
20 import com.google.common.util.concurrent.ListenableFuture;
21 import com.google.common.util.concurrent.MoreExecutors;
22 import com.google.common.util.concurrent.SettableFuture;
23 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
24 import java.io.FileOutputStream;
25 import java.io.IOException;
26 import java.util.AbstractMap.SimpleEntry;
27 import java.util.ArrayList;
28 import java.util.HashMap;
29 import java.util.List;
31 import java.util.Map.Entry;
33 import java.util.concurrent.TimeUnit;
34 import java.util.function.Function;
35 import org.apache.commons.lang3.SerializationUtils;
36 import org.eclipse.jdt.annotation.NonNull;
37 import org.opendaylight.controller.cluster.access.concepts.MemberName;
38 import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
39 import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
40 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
41 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
42 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
43 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
44 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
45 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
46 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
47 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
48 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
49 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
50 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
51 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
52 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
53 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
54 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutput;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutputBuilder;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardInput;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutput;
86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutputBuilder;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutput;
95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutputBuilder;
96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
97 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
98 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
99 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LeaderActorRefBuilder;
100 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LocalBuilder;
101 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
102 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
103 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
104 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
105 import org.opendaylight.yangtools.yang.common.Empty;
106 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
107 import org.opendaylight.yangtools.yang.common.RpcResult;
108 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
109 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
110 import org.slf4j.Logger;
111 import org.slf4j.LoggerFactory;
114 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
116 * @author Thomas Pantelis
118 public class ClusterAdminRpcService implements ClusterAdminService {
119 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
121 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
122 private static final @NonNull RpcResult<LocateShardOutput> LOCAL_SHARD_RESULT =
123 RpcResultBuilder.success(new LocateShardOutputBuilder()
124 .setMemberNode(new LocalBuilder().setLocal(Empty.getInstance()).build())
128 private final DistributedDataStoreInterface configDataStore;
129 private final DistributedDataStoreInterface operDataStore;
130 private final BindingNormalizedNodeSerializer serializer;
131 private final Timeout makeLeaderLocalTimeout;
133 public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
134 final DistributedDataStoreInterface operDataStore,
135 final BindingNormalizedNodeSerializer serializer) {
136 this.configDataStore = configDataStore;
137 this.operDataStore = operDataStore;
138 this.serializer = serializer;
140 this.makeLeaderLocalTimeout =
141 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
142 .getShardLeaderElectionTimeout().duration().$times(2));
146 public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
147 final String shardName = input.getShardName();
148 if (Strings.isNullOrEmpty(shardName)) {
149 return newFailedRpcResultFuture("A valid shard name must be specified");
152 DataStoreType dataStoreType = input.getDataStoreType();
153 if (dataStoreType == null) {
154 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
157 LOG.info("Adding replica for shard {}", shardName);
159 final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
160 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
161 Futures.addCallback(future, new FutureCallback<Success>() {
163 public void onSuccess(final Success success) {
164 LOG.info("Successfully added replica for shard {}", shardName);
165 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
169 public void onFailure(final Throwable failure) {
170 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
171 returnFuture, failure);
173 }, MoreExecutors.directExecutor());
179 public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
180 final RemoveShardReplicaInput input) {
181 final String shardName = input.getShardName();
182 if (Strings.isNullOrEmpty(shardName)) {
183 return newFailedRpcResultFuture("A valid shard name must be specified");
186 DataStoreType dataStoreType = input.getDataStoreType();
187 if (dataStoreType == null) {
188 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
191 final String memberName = input.getMemberName();
192 if (Strings.isNullOrEmpty(memberName)) {
193 return newFailedRpcResultFuture("A valid member name must be specified");
196 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
198 final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
199 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
200 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
201 Futures.addCallback(future, new FutureCallback<Success>() {
203 public void onSuccess(final Success success) {
204 LOG.info("Successfully removed replica for shard {}", shardName);
205 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
209 public void onFailure(final Throwable failure) {
210 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
211 returnFuture, failure);
213 }, MoreExecutors.directExecutor());
219 public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
220 final ActorUtils utils;
221 switch (input.getDataStoreType()) {
223 utils = configDataStore.getActorUtils();
226 utils = operDataStore.getActorUtils();
229 return newFailedRpcResultFuture("Unhandled datastore in " + input);
232 final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
233 utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
235 public void onComplete(final Throwable failure, final PrimaryShardInfo success) throws Throwable {
236 if (failure != null) {
237 LOG.debug("Failed to find shard for {}", input, failure);
238 ret.setException(failure);
242 // Data tree implies local leak
243 if (success.getLocalShardDataTree().isPresent()) {
244 ret.set(LOCAL_SHARD_RESULT);
248 final ActorSelection actorPath = success.getPrimaryShardActor();
249 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
250 .setMemberNode(new LeaderActorRefBuilder()
251 .setLeaderActorRef(actorPath.toSerializationFormat())
255 }, utils.getClientDispatcher());
261 public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
262 final String shardName = input.getShardName();
263 if (Strings.isNullOrEmpty(shardName)) {
264 return newFailedRpcResultFuture("A valid shard name must be specified");
267 DataStoreType dataStoreType = input.getDataStoreType();
268 if (dataStoreType == null) {
269 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
272 ActorUtils actorUtils = dataStoreType == DataStoreType.Config
273 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
275 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
276 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
278 final scala.concurrent.Future<ActorRef> localShardReply =
279 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<Object>() {
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<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
318 final AddPrefixShardReplicaInput input) {
320 final InstanceIdentifier<?> identifier = input.getShardPrefix();
321 if (identifier == null) {
322 return newFailedRpcResultFuture("A valid shard identifier must be specified");
325 final DataStoreType dataStoreType = input.getDataStoreType();
326 if (dataStoreType == null) {
327 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
330 LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
332 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
333 final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
334 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
335 Futures.addCallback(future, new FutureCallback<Success>() {
337 public void onSuccess(final Success success) {
338 LOG.info("Successfully added replica for shard {}", prefix);
339 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
343 public void onFailure(final Throwable failure) {
344 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
345 returnFuture, failure);
347 }, MoreExecutors.directExecutor());
353 public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
354 final RemovePrefixShardReplicaInput input) {
356 final InstanceIdentifier<?> identifier = input.getShardPrefix();
357 if (identifier == null) {
358 return newFailedRpcResultFuture("A valid shard identifier must be specified");
361 final DataStoreType dataStoreType = input.getDataStoreType();
362 if (dataStoreType == null) {
363 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
366 final String memberName = input.getMemberName();
367 if (Strings.isNullOrEmpty(memberName)) {
368 return newFailedRpcResultFuture("A valid member name must be specified");
371 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
372 identifier, memberName, dataStoreType);
373 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
375 final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
376 final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
377 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
378 Futures.addCallback(future, new FutureCallback<Success>() {
380 public void onSuccess(final Success success) {
381 LOG.info("Successfully removed replica for shard {}", prefix);
382 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
386 public void onFailure(final Throwable failure) {
387 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
388 returnFuture, failure);
390 }, MoreExecutors.directExecutor());
396 public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
397 final AddReplicasForAllShardsInput input) {
398 LOG.info("Adding replicas for all shards");
400 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
402 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
403 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
405 return waitForShardResults(shardResultData, shardResults ->
406 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
407 "Failed to add replica");
412 public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
413 final RemoveAllShardReplicasInput input) {
414 LOG.info("Removing replicas for all shards");
416 final String memberName = input.getMemberName();
417 if (Strings.isNullOrEmpty(memberName)) {
418 return newFailedRpcResultFuture("A valid member name must be specified");
421 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
422 Function<String, Object> messageSupplier = shardName ->
423 new RemoveShardReplica(shardName, MemberName.forName(memberName));
425 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
426 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
428 return waitForShardResults(shardResultData, shardResults ->
429 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
430 " Failed to remove replica");
434 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
435 final ChangeMemberVotingStatesForShardInput input) {
436 final String shardName = input.getShardName();
437 if (Strings.isNullOrEmpty(shardName)) {
438 return newFailedRpcResultFuture("A valid shard name must be specified");
441 DataStoreType dataStoreType = input.getDataStoreType();
442 if (dataStoreType == null) {
443 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
446 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
447 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
448 return newFailedRpcResultFuture("No member voting state input was specified");
451 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
454 LOG.info("Change member voting states for shard {}: {}", shardName,
455 changeVotingStatus.getMeberVotingStatusMap());
457 final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
458 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
459 Futures.addCallback(future, new FutureCallback<Success>() {
461 public void onSuccess(final Success success) {
462 LOG.info("Successfully changed member voting states for shard {}", shardName);
463 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
467 public void onFailure(final Throwable failure) {
468 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
469 returnFuture, failure);
471 }, MoreExecutors.directExecutor());
477 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
478 final ChangeMemberVotingStatesForAllShardsInput input) {
479 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
480 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
481 return newFailedRpcResultFuture("No member voting state input was specified");
484 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
485 Function<String, Object> messageSupplier = shardName ->
486 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
488 LOG.info("Change member voting states for all shards");
490 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
491 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
493 return waitForShardResults(shardResultData, shardResults ->
494 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
495 "Failed to change member voting states");
499 public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
500 final FlipMemberVotingStatesForAllShardsInput input) {
501 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
502 Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
504 LOG.info("Flip member voting states for all shards");
506 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
507 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
509 return waitForShardResults(shardResultData, shardResults ->
510 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
511 "Failed to change member voting states");
515 public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
516 final String shardName = input.getShardName();
517 if (Strings.isNullOrEmpty(shardName)) {
518 return newFailedRpcResultFuture("A valid shard name must be specified");
521 DataStoreType dataStoreType = input.getDataStoreType();
522 if (dataStoreType == null) {
523 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
526 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
528 final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
529 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
530 new GetShardRole(shardName));
531 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
533 public void onSuccess(final GetShardRoleReply reply) {
535 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
536 "No Shard role present. Please retry..").build());
539 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
540 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
541 if (reply.getRole() != null) {
542 builder.setRole(reply.getRole());
544 returnFuture.set(newSuccessfulResult(builder.build()));
548 public void onFailure(final Throwable failure) {
549 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
550 "Failed to get shard role.", failure).build());
552 }, MoreExecutors.directExecutor());
558 public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
559 final GetPrefixShardRoleInput input) {
560 final InstanceIdentifier<?> identifier = input.getShardPrefix();
561 if (identifier == null) {
562 return newFailedRpcResultFuture("A valid shard identifier must be specified");
565 final DataStoreType dataStoreType = input.getDataStoreType();
566 if (dataStoreType == null) {
567 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
570 LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
572 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
573 final String shardName = ClusterUtils.getCleanShardName(prefix);
574 final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
575 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
576 new GetShardRole(shardName));
577 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
579 public void onSuccess(final GetShardRoleReply reply) {
581 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
582 "No Shard role present. Please retry..").build());
586 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
587 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
588 if (reply.getRole() != null) {
589 builder.setRole(reply.getRole());
591 returnFuture.set(newSuccessfulResult(builder.build()));
595 public void onFailure(final Throwable failure) {
596 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
597 "Failed to get shard role.", failure).build());
599 }, MoreExecutors.directExecutor());
605 public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
606 LOG.debug("backupDatastore: {}", input);
608 if (Strings.isNullOrEmpty(input.getFilePath())) {
609 return newFailedRpcResultFuture("A valid file path must be specified");
612 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
613 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
614 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
616 public void onSuccess(final List<DatastoreSnapshot> snapshots) {
617 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
621 public void onFailure(final Throwable failure) {
622 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
624 }, MoreExecutors.directExecutor());
629 private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
630 final List<MemberVotingState> memberVotingStatus) {
631 Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
632 for (MemberVotingState memberStatus: memberVotingStatus) {
633 serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
635 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
638 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
639 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
640 final Function<List<ShardResult>, T> resultDataSupplier,
641 final String failureLogMsgPrefix) {
642 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
643 final List<ShardResult> shardResults = new ArrayList<>();
644 for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
645 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
647 public void onSuccess(final Success result) {
648 synchronized (shardResults) {
649 ShardResultBuilder shardResult = entry.getValue();
650 LOG.debug("onSuccess for shard {}, type {}", shardResult.getShardName(),
651 shardResult.getDataStoreType());
652 shardResults.add(shardResult.setSucceeded(true).build());
658 public void onFailure(final Throwable failure) {
659 synchronized (shardResults) {
660 ShardResultBuilder shardResult = entry.getValue();
661 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, shardResult.getShardName(),
662 shardResult.getDataStoreType(), failure);
663 shardResults.add(shardResult.setSucceeded(false).setErrorMessage(
664 Throwables.getRootCause(failure).getMessage()).build());
669 void checkIfComplete() {
670 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
671 if (shardResults.size() == shardResultData.size()) {
672 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
675 }, MoreExecutors.directExecutor());
680 private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
681 final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
682 final Function<String, Object> messageSupplier) {
683 ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
684 : operDataStore.getActorUtils();
685 Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
687 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
689 for (String shardName: allShardNames) {
690 ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
692 shardResultData.add(new SimpleEntry<>(future,
693 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
697 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
698 Timeout timeout = SHARD_MGR_TIMEOUT;
699 ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
700 ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
702 return Futures.allAsList(configFuture, operFuture);
705 private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
706 ActorRef shardManager = dataStoreType == DataStoreType.Config
707 ? configDataStore.getActorUtils().getShardManager()
708 : operDataStore.getActorUtils().getShardManager();
709 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
712 @SuppressWarnings("checkstyle:IllegalCatch")
713 private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
714 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
715 try (FileOutputStream fos = new FileOutputStream(fileName)) {
716 SerializationUtils.serialize(snapshots, fos);
718 returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
719 LOG.info("Successfully backed up datastore to file {}", fileName);
720 } catch (IOException | RuntimeException e) {
721 onDatastoreBackupFailure(fileName, returnFuture, e);
725 private static <T> void onDatastoreBackupFailure(final String fileName,
726 final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
727 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
730 @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
731 private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
732 final Throwable failure) {
733 LOG.error("{}", msg, failure);
734 returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
735 failure.getMessage())).build());
738 private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
739 final SettableFuture<T> returnFuture = SettableFuture.create();
741 @SuppressWarnings("unchecked")
742 scala.concurrent.Future<T> askFuture = (scala.concurrent.Future<T>) Patterns.ask(actor, message, timeout);
743 askFuture.onComplete(new OnComplete<T>() {
745 public void onComplete(final Throwable failure, final T resp) {
746 if (failure != null) {
747 returnFuture.setException(failure);
749 returnFuture.set(resp);
752 }, configDataStore.getActorUtils().getClientDispatcher());
757 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
758 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
761 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
762 return newFailedRpcResultBuilder(message, null);
765 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
766 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
769 private static <T> RpcResult<T> newSuccessfulResult(final T data) {
770 return RpcResultBuilder.success(data).build();