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;
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.MemberDownEvent;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberReachableEvent;
import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberUnreachableEvent;
.onMessage(MemberDownEvent.class, this::onPeerDown)
.onMessage(MemberReachableEvent.class, this::onPeerReachable)
.onMessage(MemberUnreachableEvent.class, this::onPeerUnreachable)
+ .onMessage(GetEntitiesRequest.class, this::onGetEntities)
+ .onMessage(GetEntityRequest.class, this::onGetEntity)
+ .onMessage(GetEntityOwnerRequest.class, this::onGetEntityOwner)
.build();
}
return;
}
- final Set<String> currentlyPresent = currentCandidates.getOrDefault(entity, Collections.emptySet());
+ final Set<String> currentlyPresent = currentCandidates.getOrDefault(entity, Set.of());
final Set<String> difference = ImmutableSet.copyOf(Sets.symmetricDifference(currentlyPresent, candidates));
LOG.debug("currently present candidates: {}", currentlyPresent);
final BiPredicate<DOMEntity, String> predicate) {
LOG.debug("Reassigning owners for {}", entities);
for (final DOMEntity entity : entities) {
-
-
if (predicate.test(entity, oldOwner)) {
ownerToEntity.remove(oldOwner, entity);
assignOwnerFor(entity);
}
private boolean isCandidateFor(final DOMEntity entity, final String candidate) {
- return currentCandidates.getOrDefault(entity, Collections.emptySet()).contains(candidate);
+ return currentCandidates.getOrDefault(entity, Set.of()).contains(candidate);
}
private void assignOwnerFor(final DOMEntity entity) {
return this;
}
+ private Behavior<OwnerSupervisorCommand> onGetEntities(final GetEntitiesRequest request) {
+ request.getReplyTo().tell(new GetEntitiesReply(currentOwners, currentCandidates));
+ return this;
+ }
+
+ private Behavior<OwnerSupervisorCommand> onGetEntity(final GetEntityRequest request) {
+ final DOMEntity entity = extractEntity(request);
+ request.getReplyTo().tell(new GetEntityReply(currentOwners.get(entity), currentCandidates.get(entity)));
+ return this;
+ }
+
+ private Behavior<OwnerSupervisorCommand> onGetEntityOwner(final GetEntityOwnerRequest request) {
+ request.getReplyTo().tell(new GetEntityOwnerReply(currentOwners.get(extractEntity(request))));
+ return this;
+ }
+
private void handleReachableEvent(final Set<String> roles) {
if (roles.contains(dataCenter)) {
activeMembers.add(extractRole(roles));
}
private Set<String> getActiveMembers() {
- final Set<String> 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));
- }
- });
+ final CurrentClusterState clusterState = cluster.state();
+ final Set<String> 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());
+ }
- return members;
+ private static DOMEntity extractEntity(final AbstractEntityRequest<?> request) {
+ return new DOMEntity(request.getType().getValue(), request.getName().getValue());
}
private static String extractRole(final Member member) {
.findFirst().orElseThrow(() -> 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."));
}