package org.opendaylight.controller.cluster.datastore.admin;
import akka.actor.ActorRef;
+import akka.actor.ActorSelection;
import akka.actor.Status.Success;
import akka.dispatch.OnComplete;
import akka.pattern.Patterns;
import akka.util.Timeout;
-import com.google.common.base.Function;
import com.google.common.base.Strings;
import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
import org.apache.commons.lang3.SerializationUtils;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.controller.cluster.access.concepts.MemberName;
import org.opendaylight.controller.cluster.datastore.DistributedDataStoreInterface;
import org.opendaylight.controller.cluster.datastore.messages.AddPrefixShardReplica;
import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
import org.opendaylight.controller.cluster.datastore.messages.ChangeShardMembersVotingStatus;
import org.opendaylight.controller.cluster.datastore.messages.FlipShardMembersVotingStatus;
+import org.opendaylight.controller.cluster.datastore.messages.GetKnownClients;
+import org.opendaylight.controller.cluster.datastore.messages.GetKnownClientsReply;
import org.opendaylight.controller.cluster.datastore.messages.GetShardRole;
import org.opendaylight.controller.cluster.datastore.messages.GetShardRoleReply;
import org.opendaylight.controller.cluster.datastore.messages.MakeLeaderLocal;
+import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
import org.opendaylight.controller.cluster.datastore.messages.RemovePrefixShardReplica;
import org.opendaylight.controller.cluster.datastore.messages.RemoveShardReplica;
import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.FlipMemberVotingStatesForAllShardsOutputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetKnownClientsForAllShardsOutputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetPrefixShardRoleOutputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.GetShardRoleOutputBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardInput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutput;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.LocateShardOutputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.MakeLeaderLocalOutputBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaInput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutput;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.RemoveShardReplicaOutputBuilder;
+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;
+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;
+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;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LeaderActorRefBuilder;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.locate.shard.output.member.node.LocalBuilder;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.member.voting.states.input.MemberVotingState;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResult;
import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.cluster.admin.rev151013.shard.result.output.ShardResultBuilder;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
+import org.opendaylight.yangtools.yang.common.Empty;
import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
private static final Timeout SHARD_MGR_TIMEOUT = new Timeout(1, TimeUnit.MINUTES);
private static final Logger LOG = LoggerFactory.getLogger(ClusterAdminRpcService.class);
+ private static final @NonNull RpcResult<LocateShardOutput> LOCAL_SHARD_RESULT =
+ RpcResultBuilder.success(new LocateShardOutputBuilder()
+ .setMemberNode(new LocalBuilder().setLocal(Empty.getInstance()).build())
+ .build())
+ .build();
private final DistributedDataStoreInterface configDataStore;
private final DistributedDataStoreInterface operDataStore;
return returnFuture;
}
+ @Override
+ public ListenableFuture<RpcResult<LocateShardOutput>> locateShard(final LocateShardInput input) {
+ final ActorUtils utils;
+ switch (input.getDataStoreType()) {
+ case Config:
+ utils = configDataStore.getActorUtils();
+ break;
+ case Operational:
+ utils = operDataStore.getActorUtils();
+ break;
+ default:
+ return newFailedRpcResultFuture("Unhandled datastore in " + input);
+ }
+
+ final SettableFuture<RpcResult<LocateShardOutput>> ret = SettableFuture.create();
+ utils.findPrimaryShardAsync(input.getShardName()).onComplete(new OnComplete<PrimaryShardInfo>() {
+ @Override
+ public void onComplete(final Throwable failure, final PrimaryShardInfo success) throws Throwable {
+ if (failure != null) {
+ LOG.debug("Failed to find shard for {}", input, failure);
+ ret.setException(failure);
+ return;
+ }
+
+ // Data tree implies local leak
+ if (success.getLocalShardDataTree().isPresent()) {
+ ret.set(LOCAL_SHARD_RESULT);
+ return;
+ }
+
+ final ActorSelection actorPath = success.getPrimaryShardActor();
+ ret.set(newSuccessfulResult(new LocateShardOutputBuilder()
+ .setMemberNode(new LeaderActorRefBuilder()
+ .setLeaderActorRef(actorPath.toSerializationFormat())
+ .build())
+ .build()));
+ }
+ }, utils.getClientDispatcher());
+
+ return ret;
+ }
+
@Override
public ListenableFuture<RpcResult<MakeLeaderLocalOutput>> makeLeaderLocal(final MakeLeaderLocalInput input) {
final String shardName = input.getShardName();
LOG.info("Adding replicas for all shards");
final List<Entry<ListenableFuture<Success>, ShardResultBuilder>> shardResultData = new ArrayList<>();
- Function<String, Object> messageSupplier = AddShardReplica::new;
- sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, messageSupplier);
- sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, messageSupplier);
+ sendMessageToManagerForConfiguredShards(DataStoreType.Config, shardResultData, AddShardReplica::new);
+ sendMessageToManagerForConfiguredShards(DataStoreType.Operational, shardResultData, AddShardReplica::new);
return waitForShardResults(shardResultData, shardResults ->
new AddReplicasForAllShardsOutputBuilder().setShardResult(shardResults).build(),
return returnFuture;
}
+
+ @Override
+ public ListenableFuture<RpcResult<GetKnownClientsForAllShardsOutput>> getKnownClientsForAllShards(
+ final GetKnownClientsForAllShardsInput input) {
+ final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies =
+ getAllShardLeadersClients();
+ return Futures.whenAllComplete(allShardReplies.values()).call(() -> processReplies(allShardReplies),
+ MoreExecutors.directExecutor());
+ }
+
+ private static RpcResult<GetKnownClientsForAllShardsOutput> processReplies(
+ final ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> allShardReplies) {
+ final List<ShardResult> result = new ArrayList<>(allShardReplies.size());
+ for (Entry<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> entry : allShardReplies.entrySet()) {
+ final ListenableFuture<GetKnownClientsReply> future = entry.getValue();
+ final ShardResultBuilder builder = new ShardResultBuilder()
+ .setDataStoreType(entry.getKey().getDataStoreType())
+ .setShardName(entry.getKey().getShardName());
+
+ final GetKnownClientsReply reply;
+ try {
+ reply = Futures.getDone(future);
+ } catch (ExecutionException e) {
+ LOG.debug("Shard {} failed to answer", entry.getKey(), e);
+ result.add(builder.setSucceeded(Boolean.FALSE).setErrorMessage(e.getCause().getMessage()).build());
+ continue;
+ }
+
+ result.add(builder
+ .setSucceeded(Boolean.TRUE)
+ .addAugmentation(ShardResult1.class, new ShardResult1Builder()
+ .setKnownClients(reply.getClients().stream()
+ .map(client -> new KnownClientsBuilder()
+ .setMember(client.getFrontendId().getMemberName().toYang())
+ .setType(client.getFrontendId().getClientType().toYang())
+ .setGeneration(client.getYangGeneration())
+ .build())
+ .collect(Collectors.toList()))
+ .build())
+ .build());
+ }
+
+ return RpcResultBuilder.success(new GetKnownClientsForAllShardsOutputBuilder().setShardResult(result).build())
+ .build();
+ }
+
private static ChangeShardMembersVotingStatus toChangeShardMembersVotingStatus(final String shardName,
final List<MemberVotingState> memberVotingStatus) {
Map<String, Boolean> serverVotingStatusMap = new HashMap<>();
return returnFuture;
}
+ private ImmutableMap<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> getAllShardLeadersClients() {
+ final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder =
+ ImmutableMap.builder();
+
+ addAllShardsClients(builder, DataStoreType.Config, configDataStore.getActorUtils());
+ addAllShardsClients(builder, DataStoreType.Operational, operDataStore.getActorUtils());
+
+ return builder.build();
+ }
+
+ private static void addAllShardsClients(
+ final ImmutableMap.Builder<ShardIdentifier, ListenableFuture<GetKnownClientsReply>> builder,
+ final DataStoreType type, final ActorUtils utils) {
+ for (String shardName : utils.getConfiguration().getAllShardNames()) {
+ final SettableFuture<GetKnownClientsReply> future = SettableFuture.create();
+ builder.put(new ShardIdentifier(type, shardName), future);
+
+ utils.findPrimaryShardAsync(shardName).flatMap(
+ info -> Patterns.ask(info.getPrimaryShardActor(), GetKnownClients.INSTANCE, SHARD_MGR_TIMEOUT),
+ utils.getClientDispatcher()).onComplete(new OnComplete<Object>() {
+ @Override
+ public void onComplete(final Throwable failure, final Object success) {
+ if (failure == null) {
+ future.set((GetKnownClientsReply) success);
+ } else {
+ future.setException(failure);
+ }
+ }
+ }, utils.getClientDispatcher());
+ }
+ }
+
private static <T> ListenableFuture<RpcResult<T>> newFailedRpcResultFuture(final String message) {
return ClusterAdminRpcService.<T>newFailedRpcResultBuilder(message).buildFuture();
}