2 * Copyright (c) 2021 PANTHEON.tech, s.r.o. 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.eos.akka;
10 import static org.awaitility.Awaitility.await;
11 import static org.junit.Assert.assertEquals;
13 import akka.actor.ActorSystem;
14 import akka.actor.Address;
15 import akka.actor.typed.ActorRef;
16 import akka.actor.typed.Behavior;
17 import akka.actor.typed.javadsl.Adapter;
18 import akka.actor.typed.javadsl.AskPattern;
19 import akka.actor.typed.javadsl.Behaviors;
20 import akka.cluster.ddata.LWWRegister;
21 import akka.cluster.ddata.LWWRegisterKey;
22 import akka.cluster.ddata.typed.javadsl.DistributedData;
23 import akka.cluster.ddata.typed.javadsl.Replicator;
24 import com.typesafe.config.Config;
25 import com.typesafe.config.ConfigFactory;
26 import java.time.Duration;
27 import java.util.ArrayList;
28 import java.util.HashMap;
29 import java.util.List;
32 import java.util.concurrent.CompletionStage;
33 import java.util.concurrent.ExecutionException;
34 import java.util.concurrent.TimeUnit;
35 import java.util.function.Supplier;
36 import org.opendaylight.controller.eos.akka.bootstrap.EOSMain;
37 import org.opendaylight.controller.eos.akka.bootstrap.command.BootstrapCommand;
38 import org.opendaylight.controller.eos.akka.bootstrap.command.GetRunningContext;
39 import org.opendaylight.controller.eos.akka.bootstrap.command.RunningContext;
40 import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberReachableEvent;
41 import org.opendaylight.controller.eos.akka.owner.supervisor.command.MemberUnreachableEvent;
42 import org.opendaylight.controller.eos.akka.owner.supervisor.command.OwnerSupervisorCommand;
43 import org.opendaylight.controller.eos.akka.registry.candidate.command.CandidateRegistryCommand;
44 import org.opendaylight.controller.eos.akka.registry.candidate.command.RegisterCandidate;
45 import org.opendaylight.controller.eos.akka.registry.candidate.command.UnregisterCandidate;
46 import org.opendaylight.controller.eos.akka.registry.listener.type.command.RegisterListener;
47 import org.opendaylight.controller.eos.akka.registry.listener.type.command.TypeListenerRegistryCommand;
48 import org.opendaylight.mdsal.eos.dom.api.DOMEntity;
49 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipChange;
50 import org.opendaylight.mdsal.eos.dom.api.DOMEntityOwnershipListener;
51 import org.slf4j.Logger;
52 import org.slf4j.LoggerFactory;
54 public abstract class AbstractNativeEosTest {
56 public static final DOMEntity ENTITY_1 = new DOMEntity("test-type", "entity-1");
57 public static final DOMEntity ENTITY_2 = new DOMEntity("test-type-2", "entity-2");
59 protected static final List<String> TWO_NODE_SEED_NODES =
60 List.of("akka://ClusterSystem@127.0.0.1:2550",
61 "akka://ClusterSystem@127.0.0.1:2551");
63 protected static final List<String> THREE_NODE_SEED_NODES =
64 List.of("akka://ClusterSystem@127.0.0.1:2550",
65 "akka://ClusterSystem@127.0.0.1:2551",
66 "akka://ClusterSystem@127.0.0.1:2552");
68 private static final String REMOTE_PROTOCOL = "akka";
69 private static final String PORT_PARAM = "akka.remote.artery.canonical.port";
70 private static final String ROLE_PARAM = "akka.cluster.roles";
71 private static final String SEED_NODES_PARAM = "akka.cluster.seed-nodes";
74 protected static ClusterNode startupRemote(final int port, final List<String> roles)
75 throws ExecutionException, InterruptedException {
76 return startup(port, roles, THREE_NODE_SEED_NODES);
79 protected static ClusterNode startupRemote(final int port, final List<String> roles, final List<String> seedNodes)
80 throws ExecutionException, InterruptedException {
81 return startup(port, roles, seedNodes);
84 protected static ClusterNode startup(final int port, final List<String> roles)
85 throws ExecutionException, InterruptedException {
86 return startup(port, roles, List.of());
89 protected static ClusterNode startup(final int port, final List<String> roles, final List<String> seedNodes)
90 throws ExecutionException, InterruptedException {
92 return startup(port, roles, seedNodes, AbstractNativeEosTest::rootBehavior);
95 protected static ClusterNode startup(final int port, final List<String> roles, final List<String> seedNodes,
96 final Supplier<Behavior<BootstrapCommand>> bootstrap)
97 throws ExecutionException, InterruptedException {
98 // Override the configuration
99 final Map<String, Object> overrides = new HashMap<>(4);
100 overrides.put(PORT_PARAM, port);
101 overrides.put(ROLE_PARAM, roles);
102 if (!seedNodes.isEmpty()) {
103 overrides.put(SEED_NODES_PARAM, seedNodes);
106 final Config config = ConfigFactory.parseMap(overrides).withFallback(ConfigFactory.load());
108 // Create a classic Akka system since thats what we will have in osgi
109 final akka.actor.ActorSystem system = akka.actor.ActorSystem.create("ClusterSystem", config);
110 final ActorRef<BootstrapCommand> eosBootstrap =
111 Adapter.spawn(system, bootstrap.get(), "EOSBootstrap");
113 final CompletionStage<RunningContext> ask = AskPattern.ask(eosBootstrap,
114 GetRunningContext::new,
115 Duration.ofSeconds(5),
116 Adapter.toTyped(system.scheduler()));
117 final RunningContext runningContext = ask.toCompletableFuture().get();
119 return new ClusterNode(port, roles, system, eosBootstrap, runningContext.getListenerRegistry(),
120 runningContext.getCandidateRegistry(), runningContext.getOwnerSupervisor());
123 private static Behavior<BootstrapCommand> rootBehavior() {
124 return Behaviors.setup(context -> EOSMain.create());
127 protected static void registerCandidates(final ClusterNode node, final DOMEntity entity, final String... members) {
128 final ActorRef<CandidateRegistryCommand> candidateRegistry = node.getCandidateRegistry();
129 registerCandidates(candidateRegistry, entity, members);
132 protected static void registerCandidates(final ActorRef<CandidateRegistryCommand> candidateRegistry,
133 final DOMEntity entity, final String... members) {
134 for (final String member : members) {
135 candidateRegistry.tell(new RegisterCandidate(entity, member));
139 protected static void unregisterCandidates(final ClusterNode node, final DOMEntity entity,
140 final String... members) {
141 final ActorRef<CandidateRegistryCommand> candidateRegistry = node.getCandidateRegistry();
142 for (final String member : members) {
143 candidateRegistry.tell(new UnregisterCandidate(entity, member));
147 protected static MockEntityOwnershipListener registerListener(final ClusterNode node, final DOMEntity entity) {
148 final ActorRef<TypeListenerRegistryCommand> listenerRegistry = node.getListenerRegistry();
149 final MockEntityOwnershipListener listener = new MockEntityOwnershipListener(node.getRoles().get(0));
150 listenerRegistry.tell(new RegisterListener(entity.getType(), listener));
155 protected static void reachableMember(final ClusterNode node, final String role) {
156 reachableMember(node.getOwnerSupervisor(), role);
159 protected static void reachableMember(final ActorRef<OwnerSupervisorCommand> ownerSupervisor, final String role) {
160 ownerSupervisor.tell(new MemberReachableEvent(
161 new Address(REMOTE_PROTOCOL, "ClusterSystem@127.0.0.1:2550"), Set.of(role)));
164 protected static void unreachableMember(final ClusterNode node, final String role) {
165 unreachableMember(node.getOwnerSupervisor(), role);
168 protected static void unreachableMember(final ActorRef<OwnerSupervisorCommand> ownerSupervisor, final String role) {
169 ownerSupervisor.tell(new MemberUnreachableEvent(
170 new Address(REMOTE_PROTOCOL, "ClusterSystem@127.0.0.1:2550"), Set.of(role)));
173 protected static void waitUntillOwnerPresent(final ClusterNode clusterNode, final DOMEntity entity) {
174 await().until(() -> {
175 final DistributedData distributedData = DistributedData.get(clusterNode.getActorSystem());
176 final CompletionStage<Replicator.GetResponse<LWWRegister<String>>> ask =
177 AskPattern.ask(distributedData.replicator(),
178 replyTo -> new Replicator.Get<>(
179 new LWWRegisterKey<>(entity.toString()), Replicator.readLocal(), replyTo),
180 Duration.ofSeconds(5),
181 clusterNode.getActorSystem().scheduler());
183 final Replicator.GetResponse<LWWRegister<String>> response =
184 ask.toCompletableFuture().get(5, TimeUnit.SECONDS);
186 if (response instanceof Replicator.GetSuccess) {
187 final String owner = ((Replicator.GetSuccess<LWWRegister<String>>) response).dataValue().getValue();
188 return !owner.isEmpty();
195 protected static void verifyListenerState(final MockEntityOwnershipListener listener, final DOMEntity entity,
196 final boolean hasOwner, final boolean isOwner, final boolean wasOwner) {
197 await().until(() -> !listener.getChanges().isEmpty());
199 await().untilAsserted(() -> {
200 final List<DOMEntityOwnershipChange> changes = listener.getChanges();
201 final DOMEntityOwnershipChange domEntityOwnershipChange = listener.getChanges().get(changes.size() - 1);
202 assertEquals(entity, domEntityOwnershipChange.getEntity());
204 assertEquals(hasOwner, domEntityOwnershipChange.getState().hasOwner());
205 assertEquals(isOwner, domEntityOwnershipChange.getState().isOwner());
206 assertEquals(wasOwner, domEntityOwnershipChange.getState().wasOwner());
210 protected static void verifyNoNotifications(final MockEntityOwnershipListener listener) {
211 await().pollDelay(2, TimeUnit.SECONDS).until(() -> listener.getChanges().isEmpty());
214 protected static final class ClusterNode {
215 private final int port;
216 private final List<String> roles;
217 private final akka.actor.typed.ActorSystem<Void> actorSystem;
218 private final ActorRef<BootstrapCommand> eosBootstrap;
219 private final ActorRef<TypeListenerRegistryCommand> listenerRegistry;
220 private final ActorRef<CandidateRegistryCommand> candidateRegistry;
221 private final ActorRef<OwnerSupervisorCommand> ownerSupervisor;
223 private ClusterNode(final int port,
224 final List<String> roles,
225 final ActorSystem actorSystem,
226 final ActorRef<BootstrapCommand> eosBootstrap,
227 final ActorRef<TypeListenerRegistryCommand> listenerRegistry,
228 final ActorRef<CandidateRegistryCommand> candidateRegistry,
229 final ActorRef<OwnerSupervisorCommand> ownerSupervisor) {
232 this.actorSystem = Adapter.toTyped(actorSystem);
233 this.eosBootstrap = eosBootstrap;
234 this.listenerRegistry = listenerRegistry;
235 this.candidateRegistry = candidateRegistry;
236 this.ownerSupervisor = ownerSupervisor;
239 public int getPort() {
243 public akka.actor.typed.ActorSystem<Void> getActorSystem() {
247 public ActorRef<BootstrapCommand> getEosBootstrap() {
251 public ActorRef<TypeListenerRegistryCommand> getListenerRegistry() {
252 return listenerRegistry;
255 public ActorRef<CandidateRegistryCommand> getCandidateRegistry() {
256 return candidateRegistry;
259 public ActorRef<OwnerSupervisorCommand> getOwnerSupervisor() {
260 return ownerSupervisor;
263 public List<String> getRoles() {
268 protected static final class MockEntityOwnershipListener implements DOMEntityOwnershipListener {
270 private final Logger log = LoggerFactory.getLogger(MockEntityOwnershipListener.class);
272 private final List<DOMEntityOwnershipChange> changes = new ArrayList<>();
273 private final String member;
275 public MockEntityOwnershipListener(final String member) {
277 this.member = member;
281 public void ownershipChanged(final DOMEntityOwnershipChange ownershipChange) {
282 log.info("{} Received ownershipCHanged: {}", member, ownershipChange);
283 log.info("{} changes: {}", member, changes.size());
284 changes.add(ownershipChange);
287 public List<DOMEntityOwnershipChange> getChanges() {