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.AddPrefixShardReplica;
44 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
45 import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
46 import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
47 import org.opendaylight.controller.cluster.datastore.messages.GetKnownClients;
48 import org.opendaylight.controller.cluster.datastore.messages.GetKnownClientsReply;
49 import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
50 import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
51 import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
52 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
53 import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
54 import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
55 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
56 import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshotList;
57 import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
58 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
59 import org.opendaylight.controller.cluster.raft.client.messages.GetSnapshot;
60 import org.opendaylight.mdsal.binding.dom.codec.api.BindingNormalizedNodeSerializer;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaInput;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutput;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddPrefixShardReplicaOutputBuilder;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsInput;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutput;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddReplicasForAllShardsOutputBuilder;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaInput;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutput;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.AddShardReplicaOutputBuilder;
70 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreInput;
71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutput;
72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.BackupDatastoreOutputBuilder;
73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsInput;
74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutput;
75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForAllShardsOutputBuilder;
76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardInput;
77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutput;
78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ChangeMemberVotingStatesForShardOutputBuilder;
79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.ClusterAdminService;
80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.DataStoreType;
81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsInput;
85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutput;
86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutputBuilder;
87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardInput;
94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutput;
95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutputBuilder;
96 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
97 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
98 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
99 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasInput;
100 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutput;
101 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveAllShardReplicasOutputBuilder;
102 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaInput;
103 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutput;
104 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemovePrefixShardReplicaOutputBuilder;
105 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
106 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
107 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
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.ShardResult1;
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.ShardResult1Builder;
110 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;
111 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;
112 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LeaderActorRefBuilder;
113 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LocalBuilder;
114 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
115 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
116 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
117 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultKey;
118 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
119 import org.opendaylight.yangtools.yang.common.Empty;
120 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
121 import org.opendaylight.yangtools.yang.common.RpcResult;
122 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
123 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
124 import org.slf4j.Logger;
125 import org.slf4j.LoggerFactory;
126 import scala.concurrent.Future;
129 * Implements the yang RPCs defined in the generated ClusterAdminService interface.
131 * @author Thomas Pantelis
133 public class ClusterAdminRpcService implements ClusterAdminService {
134 private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
136 private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
137 private static final @NonNull RpcResult<LocateShardOutput> LOCAL_SHARD_RESULT =
138 RpcResultBuilder.success(new LocateShardOutputBuilder()
139 .setMemberNode(new LocalBuilder().setLocal(Empty.getInstance()).build())
143 private final DistributedDataStoreInterface configDataStore;
144 private final DistributedDataStoreInterface operDataStore;
145 private final BindingNormalizedNodeSerializer serializer;
146 private final Timeout makeLeaderLocalTimeout;
148 public ClusterAdminRpcService(final DistributedDataStoreInterface configDataStore,
149 final DistributedDataStoreInterface operDataStore,
150 final BindingNormalizedNodeSerializer serializer) {
151 this.configDataStore = configDataStore;
152 this.operDataStore = operDataStore;
153 this.serializer = serializer;
155 this.makeLeaderLocalTimeout =
156 new Timeout(configDataStore.getActorUtils().getDatastoreContext()
157 .getShardLeaderElectionTimeout().duration().$times(2));
161 public ListenableFuture<RpcResult<AddShardReplicaOutput>> addShardReplica(final AddShardReplicaInput input) {
162 final String shardName = input.getShardName();
163 if (Strings.isNullOrEmpty(shardName)) {
164 return newFailedRpcResultFuture("A valid shard name must be specified");
167 DataStoreType dataStoreType = input.getDataStoreType();
168 if (dataStoreType == null) {
169 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
172 LOG.info("Adding replica for shard {}", shardName);
174 final SettableFuture<RpcResult<AddShardReplicaOutput>> returnFuture = SettableFuture.create();
175 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddShardReplica(shardName));
176 Futures.addCallback(future, new FutureCallback<Success>() {
178 public void onSuccess(final Success success) {
179 LOG.info("Successfully added replica for shard {}", shardName);
180 returnFuture.set(newSuccessfulResult(new AddShardReplicaOutputBuilder().build()));
184 public void onFailure(final Throwable failure) {
185 onMessageFailure(String.format("Failed to add replica for shard %s", shardName),
186 returnFuture, failure);
188 }, MoreExecutors.directExecutor());
194 public ListenableFuture<RpcResult<RemoveShardReplicaOutput>> removeShardReplica(
195 final RemoveShardReplicaInput input) {
196 final String shardName = input.getShardName();
197 if (Strings.isNullOrEmpty(shardName)) {
198 return newFailedRpcResultFuture("A valid shard name must be specified");
201 DataStoreType dataStoreType = input.getDataStoreType();
202 if (dataStoreType == null) {
203 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
206 final String memberName = input.getMemberName();
207 if (Strings.isNullOrEmpty(memberName)) {
208 return newFailedRpcResultFuture("A valid member name must be specified");
211 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}", shardName, memberName, dataStoreType);
213 final SettableFuture<RpcResult<RemoveShardReplicaOutput>> returnFuture = SettableFuture.create();
214 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
215 new RemoveShardReplica(shardName, MemberName.forName(memberName)));
216 Futures.addCallback(future, new FutureCallback<Success>() {
218 public void onSuccess(final Success success) {
219 LOG.info("Successfully removed replica for shard {}", shardName);
220 returnFuture.set(newSuccessfulResult(new RemoveShardReplicaOutputBuilder().build()));
224 public void onFailure(final Throwable failure) {
225 onMessageFailure(String.format("Failed to remove replica for shard %s", shardName),
226 returnFuture, failure);
228 }, MoreExecutors.directExecutor());
234 public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
235 final ActorUtils utils;
236 switch (input.getDataStoreType()) {
238 utils = configDataStore.getActorUtils();
241 utils = operDataStore.getActorUtils();
244 return newFailedRpcResultFuture("Unhandled datastore in " + input);
247 final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
248 utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
250 public void onComplete(final Throwable failure, final PrimaryShardInfo success) throws Throwable {
251 if (failure != null) {
252 LOG.debug("Failed to find shard for {}", input, failure);
253 ret.setException(failure);
257 // Data tree implies local leak
258 if (success.getLocalShardDataTree().isPresent()) {
259 ret.set(LOCAL_SHARD_RESULT);
263 final ActorSelection actorPath = success.getPrimaryShardActor();
264 ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
265 .setMemberNode(new LeaderActorRefBuilder()
266 .setLeaderActorRef(actorPath.toSerializationFormat())
270 }, utils.getClientDispatcher());
276 public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
277 final String shardName = input.getShardName();
278 if (Strings.isNullOrEmpty(shardName)) {
279 return newFailedRpcResultFuture("A valid shard name must be specified");
282 DataStoreType dataStoreType = input.getDataStoreType();
283 if (dataStoreType == null) {
284 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
287 ActorUtils actorUtils = dataStoreType == DataStoreType.Config
288 ? configDataStore.getActorUtils() : operDataStore.getActorUtils();
290 LOG.info("Moving leader to local node {} for shard {}, datastoreType {}",
291 actorUtils.getCurrentMemberName().getName(), shardName, dataStoreType);
293 final Future<ActorRef> localShardReply = actorUtils.findLocalShardAsync(shardName);
295 final scala.concurrent.Promise<Object> makeLeaderLocalAsk = akka.dispatch.Futures.promise();
296 localShardReply.onComplete(new OnComplete<ActorRef>() {
298 public void onComplete(final Throwable failure, final ActorRef actorRef) {
299 if (failure != null) {
300 LOG.warn("No local shard found for {} datastoreType {} - Cannot request leadership transfer to"
301 + " local shard.", shardName, dataStoreType, failure);
302 makeLeaderLocalAsk.failure(failure);
305 .completeWith(actorUtils
306 .executeOperationAsync(actorRef, MakeLeaderLocal.INSTANCE, makeLeaderLocalTimeout));
309 }, actorUtils.getClientDispatcher());
311 final SettableFuture<RpcResult<MakeLeaderLocalOutput>> future = SettableFuture.create();
312 makeLeaderLocalAsk.future().onComplete(new OnComplete<>() {
314 public void onComplete(final Throwable failure, final Object success) {
315 if (failure != null) {
316 LOG.error("Leadership transfer failed for shard {}.", shardName, failure);
317 future.set(RpcResultBuilder.<MakeLeaderLocalOutput>failed().withError(ErrorType.APPLICATION,
318 "leadership transfer failed", failure).build());
322 LOG.debug("Leadership transfer complete");
323 future.set(RpcResultBuilder.success(new MakeLeaderLocalOutputBuilder().build()).build());
325 }, actorUtils.getClientDispatcher());
331 public ListenableFuture<RpcResult<AddPrefixShardReplicaOutput>> addPrefixShardReplica(
332 final AddPrefixShardReplicaInput input) {
334 final InstanceIdentifier<?> identifier = input.getShardPrefix();
335 if (identifier == null) {
336 return newFailedRpcResultFuture("A valid shard identifier must be specified");
339 final DataStoreType dataStoreType = input.getDataStoreType();
340 if (dataStoreType == null) {
341 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
344 LOG.info("Adding replica for shard {}, datastore type {}", identifier, dataStoreType);
346 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
347 final SettableFuture<RpcResult<AddPrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
348 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, new AddPrefixShardReplica(prefix));
349 Futures.addCallback(future, new FutureCallback<Success>() {
351 public void onSuccess(final Success success) {
352 LOG.info("Successfully added replica for shard {}", prefix);
353 returnFuture.set(newSuccessfulResult(new AddPrefixShardReplicaOutputBuilder().build()));
357 public void onFailure(final Throwable failure) {
358 onMessageFailure(String.format("Failed to add replica for shard %s", prefix),
359 returnFuture, failure);
361 }, MoreExecutors.directExecutor());
367 public ListenableFuture<RpcResult<RemovePrefixShardReplicaOutput>> removePrefixShardReplica(
368 final RemovePrefixShardReplicaInput input) {
370 final InstanceIdentifier<?> identifier = input.getShardPrefix();
371 if (identifier == null) {
372 return newFailedRpcResultFuture("A valid shard identifier must be specified");
375 final DataStoreType dataStoreType = input.getDataStoreType();
376 if (dataStoreType == null) {
377 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
380 final String memberName = input.getMemberName();
381 if (Strings.isNullOrEmpty(memberName)) {
382 return newFailedRpcResultFuture("A valid member name must be specified");
385 LOG.info("Removing replica for shard {} memberName {}, datastoreType {}",
386 identifier, memberName, dataStoreType);
387 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
389 final SettableFuture<RpcResult<RemovePrefixShardReplicaOutput>> returnFuture = SettableFuture.create();
390 final ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType,
391 new RemovePrefixShardReplica(prefix, MemberName.forName(memberName)));
392 Futures.addCallback(future, new FutureCallback<Success>() {
394 public void onSuccess(final Success success) {
395 LOG.info("Successfully removed replica for shard {}", prefix);
396 returnFuture.set(newSuccessfulResult(new RemovePrefixShardReplicaOutputBuilder().build()));
400 public void onFailure(final Throwable failure) {
401 onMessageFailure(String.format("Failed to remove replica for shard %s", prefix),
402 returnFuture, failure);
404 }, MoreExecutors.directExecutor());
410 public ListenableFuture<RpcResult<AddReplicasForAllShardsOutput>> addReplicasForAllShards(
411 final AddReplicasForAllShardsInput input) {
412 LOG.info("Adding replicas for all shards");
414 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
416 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
417 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
419 return waitForShardResults(shardResultData, shardResults ->
420 new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
421 "Failed to add replica");
426 public ListenableFuture<RpcResult<RemoveAllShardReplicasOutput>> removeAllShardReplicas(
427 final RemoveAllShardReplicasInput input) {
428 LOG.info("Removing replicas for all shards");
430 final String memberName = input.getMemberName();
431 if (Strings.isNullOrEmpty(memberName)) {
432 return newFailedRpcResultFuture("A valid member name must be specified");
435 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
436 Function<String, Object> messageSupplier = shardName ->
437 new RemoveShardReplica(shardName, MemberName.forName(memberName));
439 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
440 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
442 return waitForShardResults(shardResultData, shardResults ->
443 new RemoveAllShardReplicasOutputBuilder().setShardResult(shardResults).build(),
444 " Failed to remove replica");
448 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> changeMemberVotingStatesForShard(
449 final ChangeMemberVotingStatesForShardInput input) {
450 final String shardName = input.getShardName();
451 if (Strings.isNullOrEmpty(shardName)) {
452 return newFailedRpcResultFuture("A valid shard name must be specified");
455 DataStoreType dataStoreType = input.getDataStoreType();
456 if (dataStoreType == null) {
457 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
460 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
461 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
462 return newFailedRpcResultFuture("No member voting state input was specified");
465 ChangeShardMembersVotingStatus changeVotingStatus = toChangeShardMembersVotingStatus(shardName,
468 LOG.info("Change member voting states for shard {}: {}", shardName,
469 changeVotingStatus.getMeberVotingStatusMap());
471 final SettableFuture<RpcResult<ChangeMemberVotingStatesForShardOutput>> returnFuture = SettableFuture.create();
472 ListenableFuture<Success> future = sendMessageToShardManager(dataStoreType, changeVotingStatus);
473 Futures.addCallback(future, new FutureCallback<Success>() {
475 public void onSuccess(final Success success) {
476 LOG.info("Successfully changed member voting states for shard {}", shardName);
477 returnFuture.set(newSuccessfulResult(new ChangeMemberVotingStatesForShardOutputBuilder().build()));
481 public void onFailure(final Throwable failure) {
482 onMessageFailure(String.format("Failed to change member voting states for shard %s", shardName),
483 returnFuture, failure);
485 }, MoreExecutors.directExecutor());
491 public ListenableFuture<RpcResult<ChangeMemberVotingStatesForAllShardsOutput>> changeMemberVotingStatesForAllShards(
492 final ChangeMemberVotingStatesForAllShardsInput input) {
493 List<MemberVotingState> memberVotingStates = input.getMemberVotingState();
494 if (memberVotingStates == null || memberVotingStates.isEmpty()) {
495 return newFailedRpcResultFuture("No member voting state input was specified");
498 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
499 Function<String, Object> messageSupplier = shardName ->
500 toChangeShardMembersVotingStatus(shardName, memberVotingStates);
502 LOG.info("Change member voting states for all shards");
504 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
505 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
507 return waitForShardResults(shardResultData, shardResults ->
508 new ChangeMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
509 "Failed to change member voting states");
513 public ListenableFuture<RpcResult<FlipMemberVotingStatesForAllShardsOutput>> flipMemberVotingStatesForAllShards(
514 final FlipMemberVotingStatesForAllShardsInput input) {
515 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
516 Function<String, Object> messageSupplier = FlipShardMembersVotingStatus::new;
518 LOG.info("Flip member voting states for all shards");
520 sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
521 sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
523 return waitForShardResults(shardResultData, shardResults ->
524 new FlipMemberVotingStatesForAllShardsOutputBuilder().setShardResult(shardResults).build(),
525 "Failed to change member voting states");
529 public ListenableFuture<RpcResult<GetShardRoleOutput>> getShardRole(final GetShardRoleInput input) {
530 final String shardName = input.getShardName();
531 if (Strings.isNullOrEmpty(shardName)) {
532 return newFailedRpcResultFuture("A valid shard name must be specified");
535 DataStoreType dataStoreType = input.getDataStoreType();
536 if (dataStoreType == null) {
537 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
540 LOG.info("Getting role for shard {}, datastore type {}", shardName, dataStoreType);
542 final SettableFuture<RpcResult<GetShardRoleOutput>> returnFuture = SettableFuture.create();
543 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
544 new GetShardRole(shardName));
545 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
547 public void onSuccess(final GetShardRoleReply reply) {
549 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
550 "No Shard role present. Please retry..").build());
553 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
554 final GetShardRoleOutputBuilder builder = new GetShardRoleOutputBuilder();
555 if (reply.getRole() != null) {
556 builder.setRole(reply.getRole());
558 returnFuture.set(newSuccessfulResult(builder.build()));
562 public void onFailure(final Throwable failure) {
563 returnFuture.set(ClusterAdminRpcService.<GetShardRoleOutput>newFailedRpcResultBuilder(
564 "Failed to get shard role.", failure).build());
566 }, MoreExecutors.directExecutor());
572 public ListenableFuture<RpcResult<GetPrefixShardRoleOutput>> getPrefixShardRole(
573 final GetPrefixShardRoleInput input) {
574 final InstanceIdentifier<?> identifier = input.getShardPrefix();
575 if (identifier == null) {
576 return newFailedRpcResultFuture("A valid shard identifier must be specified");
579 final DataStoreType dataStoreType = input.getDataStoreType();
580 if (dataStoreType == null) {
581 return newFailedRpcResultFuture("A valid DataStoreType must be specified");
584 LOG.info("Getting prefix shard role for shard: {}, datastore type {}", identifier, dataStoreType);
586 final YangInstanceIdentifier prefix = serializer.toYangInstanceIdentifier(identifier);
587 final String shardName = ClusterUtils.getCleanShardName(prefix);
588 final SettableFuture<RpcResult<GetPrefixShardRoleOutput>> returnFuture = SettableFuture.create();
589 ListenableFuture<GetShardRoleReply> future = sendMessageToShardManager(dataStoreType,
590 new GetShardRole(shardName));
591 Futures.addCallback(future, new FutureCallback<GetShardRoleReply>() {
593 public void onSuccess(final GetShardRoleReply reply) {
595 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
596 "No Shard role present. Please retry..").build());
600 LOG.info("Successfully received role:{} for shard {}", reply.getRole(), shardName);
601 final GetPrefixShardRoleOutputBuilder builder = new GetPrefixShardRoleOutputBuilder();
602 if (reply.getRole() != null) {
603 builder.setRole(reply.getRole());
605 returnFuture.set(newSuccessfulResult(builder.build()));
609 public void onFailure(final Throwable failure) {
610 returnFuture.set(ClusterAdminRpcService.<GetPrefixShardRoleOutput>newFailedRpcResultBuilder(
611 "Failed to get shard role.", failure).build());
613 }, MoreExecutors.directExecutor());
619 public ListenableFuture<RpcResult<BackupDatastoreOutput>> backupDatastore(final BackupDatastoreInput input) {
620 LOG.debug("backupDatastore: {}", input);
622 if (Strings.isNullOrEmpty(input.getFilePath())) {
623 return newFailedRpcResultFuture("A valid file path must be specified");
626 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture = SettableFuture.create();
627 ListenableFuture<List<DatastoreSnapshot>> future = sendMessageToShardManagers(GetSnapshot.INSTANCE);
628 Futures.addCallback(future, new FutureCallback<List<DatastoreSnapshot>>() {
630 public void onSuccess(final List<DatastoreSnapshot> snapshots) {
631 saveSnapshotsToFile(new DatastoreSnapshotList(snapshots), input.getFilePath(), returnFuture);
635 public void onFailure(final Throwable failure) {
636 onDatastoreBackupFailure(input.getFilePath(), returnFuture, failure);
638 }, MoreExecutors.directExecutor());
645 public ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
646 final GetKnownClientsForAllShardsInput input) {
647 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies =
648 getAllShardLeadersClients();
649 return Futures.whenAllComplete(allShardReplies.values()).call(() -> processReplies(allShardReplies),
650 MoreExecutors.directExecutor());
653 private static RpcResult<GetKnownClientsForAllShardsOutput> processReplies(
654 final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies) {
655 final Map<ShardResultKey, ShardResult> result = Maps.newHashMapWithExpectedSize(allShardReplies.size());
656 for (Entry<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> entry : allShardReplies.entrySet()) {
657 final ListenableFuture<GetKnownClientsReply> future = entry.getValue();
658 final ShardResultBuilder builder = new ShardResultBuilder()
659 .setDataStoreType(entry.getKey().getDataStoreType())
660 .setShardName(entry.getKey().getShardName());
662 final GetKnownClientsReply reply;
664 reply = Futures.getDone(future);
665 } catch (ExecutionException e) {
666 LOG.debug("Shard {} failed to answer", entry.getKey(), e);
667 final ShardResult sr = builder
668 .setSucceeded(Boolean.FALSE)
669 .setErrorMessage(e.getCause().getMessage())
671 result.put(sr.key(), sr);
675 final ShardResult sr = builder
676 .setSucceeded(Boolean.TRUE)
677 .addAugmentation(ShardResult1.class, new ShardResult1Builder()
678 .setKnownClients(reply.getClients().stream()
679 .map(client -> new KnownClientsBuilder()
680 .setMember(client.getFrontendId().getMemberName().toYang())
681 .setType(client.getFrontendId().getClientType().toYang())
682 .setGeneration(client.getYangGeneration())
684 .collect(Collectors.toMap(KnownClients::key, Function.identity())))
688 result.put(sr.key(), sr);
691 return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
695 private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
696 final List<MemberVotingState> memberVotingStatus) {
697 Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
698 for (MemberVotingState memberStatus: memberVotingStatus) {
699 serverVotingStatusMap.put(memberStatus.getMemberName(), memberStatus.isVoting());
701 return new ChangeShardMembersVotingStatus(shardName, serverVotingStatusMap);
704 private static <T> SettableFuture<RpcResult<T>> waitForShardResults(
705 final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData,
706 final Function<Map<ShardResultKey, ShardResult>, T> resultDataSupplier,
707 final String failureLogMsgPrefix) {
708 final SettableFuture<RpcResult<T>> returnFuture = SettableFuture.create();
709 final Map<ShardResultKey, ShardResult> shardResults = new HashMap<>();
710 for (final Entry<ListenableFuture<Success>, ShardResultBuilder> entry : shardResultData) {
711 Futures.addCallback(entry.getKey(), new FutureCallback<Success>() {
713 public void onSuccess(final Success result) {
714 synchronized (shardResults) {
715 final ShardResultBuilder builder = entry.getValue();
716 LOG.debug("onSuccess for shard {}, type {}", builder.getShardName(),
717 builder.getDataStoreType());
718 final ShardResult sr = builder.setSucceeded(Boolean.TRUE).build();
719 shardResults.put(sr.key(), sr);
725 public void onFailure(final Throwable failure) {
726 synchronized (shardResults) {
727 ShardResultBuilder builder = entry.getValue();
728 LOG.warn("{} for shard {}, type {}", failureLogMsgPrefix, builder.getShardName(),
729 builder.getDataStoreType(), failure);
730 final ShardResult sr = builder
731 .setSucceeded(Boolean.FALSE)
732 .setErrorMessage(Throwables.getRootCause(failure).getMessage())
734 shardResults.put(sr.key(), sr);
739 void checkIfComplete() {
740 LOG.debug("checkIfComplete: expected {}, actual {}", shardResultData.size(), shardResults.size());
741 if (shardResults.size() == shardResultData.size()) {
742 returnFuture.set(newSuccessfulResult(resultDataSupplier.apply(shardResults)));
745 }, MoreExecutors.directExecutor());
750 private <T> void sendMessageToManagerForConfiguredShards(final DataStoreType dataStoreType,
751 final List<Entry<ListenableFuture<T>, ShardResultBuilder>> shardResultData,
752 final Function<String, Object> messageSupplier) {
753 ActorUtils actorUtils = dataStoreType == DataStoreType.Config ? configDataStore.getActorUtils()
754 : operDataStore.getActorUtils();
755 Set<String> allShardNames = actorUtils.getConfiguration().getAllShardNames();
757 LOG.debug("Sending message to all shards {} for data store {}", allShardNames, actorUtils.getDataStoreName());
759 for (String shardName: allShardNames) {
760 ListenableFuture<T> future = this.ask(actorUtils.getShardManager(), messageSupplier.apply(shardName),
762 shardResultData.add(new SimpleEntry<>(future,
763 new ShardResultBuilder().setShardName(shardName).setDataStoreType(dataStoreType)));
767 private <T> ListenableFuture<List<T>> sendMessageToShardManagers(final Object message) {
768 Timeout timeout = SHARD_MGR_TIMEOUT;
769 ListenableFuture<T> configFuture = ask(configDataStore.getActorUtils().getShardManager(), message, timeout);
770 ListenableFuture<T> operFuture = ask(operDataStore.getActorUtils().getShardManager(), message, timeout);
772 return Futures.allAsList(configFuture, operFuture);
775 private <T> ListenableFuture<T> sendMessageToShardManager(final DataStoreType dataStoreType, final Object message) {
776 ActorRef shardManager = dataStoreType == DataStoreType.Config
777 ? configDataStore.getActorUtils().getShardManager()
778 : operDataStore.getActorUtils().getShardManager();
779 return ask(shardManager, message, SHARD_MGR_TIMEOUT);
782 @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
783 justification = "https://github.com/spotbugs/spotbugs/issues/811")
784 @SuppressWarnings("checkstyle:IllegalCatch")
785 private static void saveSnapshotsToFile(final DatastoreSnapshotList snapshots, final String fileName,
786 final SettableFuture<RpcResult<BackupDatastoreOutput>> returnFuture) {
787 try (FileOutputStream fos = new FileOutputStream(fileName)) {
788 SerializationUtils.serialize(snapshots, fos);
790 returnFuture.set(newSuccessfulResult(new BackupDatastoreOutputBuilder().build()));
791 LOG.info("Successfully backed up datastore to file {}", fileName);
792 } catch (IOException | RuntimeException e) {
793 onDatastoreBackupFailure(fileName, returnFuture, e);
797 private static <T> void onDatastoreBackupFailure(final String fileName,
798 final SettableFuture<RpcResult<T>> returnFuture, final Throwable failure) {
799 onMessageFailure(String.format("Failed to back up datastore to file %s", fileName), returnFuture, failure);
802 @SuppressFBWarnings("SLF4J_SIGN_ONLY_FORMAT")
803 private static <T> void onMessageFailure(final String msg, final SettableFuture<RpcResult<T>> returnFuture,
804 final Throwable failure) {
805 LOG.error("{}", msg, failure);
806 returnFuture.set(ClusterAdminRpcService.<T>newFailedRpcResultBuilder(String.format("%s: %s", msg,
807 failure.getMessage())).build());
810 private <T> ListenableFuture<T> ask(final ActorRef actor, final Object message, final Timeout timeout) {
811 final SettableFuture<T> returnFuture = SettableFuture.create();
813 @SuppressWarnings("unchecked")
814 Future<T> askFuture = (Future<T>) Patterns.ask(actor, message, timeout);
815 askFuture.onComplete(new OnComplete<T>() {
817 public void onComplete(final Throwable failure, final T resp) {
818 if (failure != null) {
819 returnFuture.setException(failure);
821 returnFuture.set(resp);
824 }, configDataStore.getActorUtils().getClientDispatcher());
829 private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
830 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
831 ImmutableMap.builder();
833 addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
834 addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
836 return builder.build();
839 private static void addAllShardsClients(
840 final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder,
841 final DataStoreType type, final ActorUtils utils) {
842 for (String shardName : utils.getConfiguration().getAllShardNames()) {
843 final SettableFuture<GetKnownClientsReply> future = SettableFuture.create();
844 builder.put(new ShardIdentifier(type, shardName), future);
846 utils.findPrimaryShardAsync(shardName).flatMap(
847 info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
848 utils.getClientDispatcher()).onComplete(new OnComplete<>() {
850 public void onComplete(final Throwable failure, final Object success) {
851 if (failure == null) {
852 future.set((GetKnownClientsReply) success);
854 future.setException(failure);
857 }, utils.getClientDispatcher());
861 private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
862 return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
865 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message) {
866 return newFailedRpcResultBuilder(message, null);
869 private static <T> RpcResultBuilder<T> newFailedRpcResultBuilder(final String message, final Throwable cause) {
870 return RpcResultBuilder.<T>failed().withError(ErrorType.RPC, message, cause);
873 private static <T> RpcResult<T> newSuccessfulResult(final T data) {
874 return RpcResultBuilder.success(data).build();