X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Feos-dom-akka%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Feos%2Fakka%2Fowner%2Fsupervisor%2FOwnerSupervisor.java;h=9841b65b7bd7e53fff42ccbef4987369822b3b59;hp=5210f43dc66cbb9ade382f301cd8c42a2ca9844f;hb=1adec580405aafae353f8b0b3a5a0f474a05c6c0;hpb=fd7bc3ff484b7b0bd72d425d59883f710e397c3f diff --git a/opendaylight/md-sal/eos-dom-akka/src/main/java/org/opendaylight/controller/eos/akka/owner/supervisor/OwnerSupervisor.java b/opendaylight/md-sal/eos-dom-akka/src/main/java/org/opendaylight/controller/eos/akka/owner/supervisor/OwnerSupervisor.java index 5210f43dc6..9841b65b7b 100644 --- a/opendaylight/md-sal/eos-dom-akka/src/main/java/org/opendaylight/controller/eos/akka/owner/supervisor/OwnerSupervisor.java +++ b/opendaylight/md-sal/eos-dom-akka/src/main/java/org/opendaylight/controller/eos/akka/owner/supervisor/OwnerSupervisor.java @@ -7,6 +7,9 @@ */ package org.opendaylight.controller.eos.akka.owner.supervisor; +import static com.google.common.base.Verify.verifyNotNull; +import static java.util.Objects.requireNonNull; + import akka.actor.typed.ActorRef; import akka.actor.typed.Behavior; import akka.actor.typed.javadsl.AbstractBehavior; @@ -14,6 +17,7 @@ import akka.actor.typed.javadsl.ActorContext; import akka.actor.typed.javadsl.Behaviors; import akka.actor.typed.javadsl.Receive; import akka.cluster.ClusterEvent; +import akka.cluster.ClusterEvent.CurrentClusterState; import akka.cluster.Member; import akka.cluster.ddata.LWWRegister; import akka.cluster.ddata.LWWRegisterKey; @@ -25,6 +29,7 @@ import akka.cluster.ddata.typed.javadsl.Replicator; import akka.cluster.ddata.typed.javadsl.ReplicatorMessageAdapter; import akka.cluster.typed.Cluster; import akka.cluster.typed.Subscribe; +import akka.pattern.StatusReply; import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableSet; @@ -33,23 +38,23 @@ import com.google.common.collect.Sets; import java.time.Duration; import java.util.ArrayList; import java.util.Collection; -import java.util.Collections; import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.BiPredicate; import java.util.stream.Collectors; +import java.util.stream.StreamSupport; import org.opendaylight.controller.eos.akka.owner.supervisor.command.AbstractEntityRequest; import org.opendaylight.controller.eos.akka.owner.supervisor.command.CandidatesChanged; import org.opendaylight.controller.eos.akka.owner.supervisor.command.DataCenterDeactivated; import org.opendaylight.controller.eos.akka.owner.supervisor.command.DeactivateDataCenter; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntitiesReply; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntitiesRequest; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityOwnerReply; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityOwnerRequest; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityReply; -import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityRequest; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntitiesBackendReply; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntitiesBackendRequest; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityBackendReply; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityBackendRequest; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityOwnerBackendReply; +import org.opendaylight.controller.eos.akka.owner.supervisor.command.GetEntityOwnerBackendRequest; import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberDownEvent; import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberReachableEvent; import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberUnreachableEvent; @@ -57,6 +62,7 @@ import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberUpEve import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerChanged; import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorCommand; import org.opendaylight.controller.eos.akka.registry.candidate.CandidateRegistry; +import org.opendaylight.mdsal.binding.dom.codec.api.BindingInstanceIdentifierCodec; import org.opendaylight.mdsal.eos.dom.api.DOMEntity; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -96,10 +102,14 @@ public final class OwnerSupervisor extends AbstractBehavior reassignPredicate = (entity, candidate) -> !isActiveCandidate(candidate) || !isCandidateFor(entity, candidate); + private final BindingInstanceIdentifierCodec iidCodec; + private OwnerSupervisor(final ActorContext context, final Map> currentCandidates, - final Map currentOwners) { + final Map currentOwners, + final BindingInstanceIdentifierCodec iidCodec) { super(context); + this.iidCodec = requireNonNull(iidCodec); final DistributedData distributedData = DistributedData.get(context.getSystem()); final ActorRef replicator = distributedData.replicator(); @@ -149,8 +159,8 @@ public final class OwnerSupervisor extends AbstractBehavior create(final Map> currentCandidates, - final Map currentOwners) { - return Behaviors.setup(ctx -> new OwnerSupervisor(ctx, currentCandidates, currentOwners)); + final Map currentOwners, final BindingInstanceIdentifierCodec iidCodec) { + return Behaviors.setup(ctx -> new OwnerSupervisor(ctx, currentCandidates, currentOwners, iidCodec)); } @Override @@ -163,16 +173,16 @@ public final class OwnerSupervisor extends AbstractBehavior onDeactivateDatacenter(final DeactivateDataCenter command) { LOG.debug("Deactivating Owner Supervisor on {}", cluster.selfMember()); command.getReplyTo().tell(DataCenterDeactivated.INSTANCE); - return IdleSupervisor.create(); + return IdleSupervisor.create(iidCodec); } private Behavior onOwnerChanged(final OwnerChanged command) { @@ -234,7 +244,7 @@ public final class OwnerSupervisor extends AbstractBehavior currentlyPresent = currentCandidates.getOrDefault(entity, Collections.emptySet()); + final Set currentlyPresent = currentCandidates.getOrDefault(entity, Set.of()); final Set difference = ImmutableSet.copyOf(Sets.symmetricDifference(currentlyPresent, candidates)); LOG.debug("currently present candidates: {}", currentlyPresent); @@ -297,7 +307,7 @@ public final class OwnerSupervisor extends AbstractBehavior onGetEntities(final GetEntitiesRequest request) { - request.getReplyTo().tell(new GetEntitiesReply(currentOwners, currentCandidates)); + private Behavior onGetEntities(final GetEntitiesBackendRequest request) { + request.getReplyTo().tell(StatusReply.success(new GetEntitiesBackendReply(currentOwners, currentCandidates))); return this; } - private Behavior onGetEntity(final GetEntityRequest request) { + private Behavior onGetEntity(final GetEntityBackendRequest request) { final DOMEntity entity = extractEntity(request); - request.getReplyTo().tell(new GetEntityReply(currentOwners.get(entity), currentCandidates.get(entity))); + request.getReplyTo().tell(StatusReply.success( + new GetEntityBackendReply(currentOwners.get(entity), currentCandidates.get(entity)))); return this; } - private Behavior onGetEntityOwner(final GetEntityOwnerRequest request) { - request.getReplyTo().tell(new GetEntityOwnerReply(currentOwners.get(extractEntity(request)))); + private Behavior onGetEntityOwner(final GetEntityOwnerBackendRequest request) { + request.getReplyTo().tell( + StatusReply.success(new GetEntityOwnerBackendReply(currentOwners.get(extractEntity(request))))); return this; } @@ -413,25 +425,31 @@ public final class OwnerSupervisor extends AbstractBehavior getActiveMembers() { - final Set members = new HashSet<>(); - cluster.state().getMembers().forEach(member -> members.add(extractRole(member))); - // filter out unreachable - members.removeAll(cluster.state().getUnreachable().stream() - .map(OwnerSupervisor::extractRole) - .collect(Collectors.toSet())); - - // filter out members not from our datacenter - cluster.state().getMembers().forEach(member -> { - if (!member.roles().contains(dataCenter)) { - members.remove(extractRole(member)); - } - }); - - return members; + final CurrentClusterState clusterState = cluster.state(); + final Set unreachableRoles = clusterState.getUnreachable().stream() + .map(OwnerSupervisor::extractRole) + .collect(Collectors.toSet()); + + return StreamSupport.stream(clusterState.getMembers().spliterator(), false) + // We are evaluating the set of roles for each member + .map(Member::getRoles) + // Filter out any members which do not share our dataCenter + .filter(roles -> roles.contains(dataCenter)) + // Find first legal role + .map(OwnerSupervisor::extractRole) + // filter out unreachable roles + .filter(role -> !unreachableRoles.contains(role)) + .collect(Collectors.toSet()); } - private static DOMEntity extractEntity(final AbstractEntityRequest request) { - return new DOMEntity(request.getType().getValue(), request.getName().getValue()); + private DOMEntity extractEntity(final AbstractEntityRequest request) { + final var name = request.getName(); + final var iid = name.getInstanceIdentifier(); + if (iid != null) { + return new DOMEntity(request.getType().getValue(), iidCodec.fromBinding(iid)); + } + final var str = verifyNotNull(name.getString(), "Unhandled entity name %s", name); + return new DOMEntity(request.getType().getValue(), str); } private static String extractRole(final Member member) { @@ -443,7 +461,7 @@ public final class OwnerSupervisor extends AbstractBehavior new IllegalArgumentException("No valid role found.")); } - private String extractDatacenterRole(final Member member) { + private static String extractDatacenterRole(final Member member) { return member.getRoles().stream().filter(role -> role.startsWith(DATACENTER_PREFIX)) .findFirst().orElseThrow(() -> new IllegalArgumentException("No valid role found.")); }