2 * Copyright (c) 2014, 2015 Cisco 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
9 package org.opendaylight.controller.remote.rpc.registry;
11 import static org.junit.Assert.fail;
13 import akka.actor.ActorRef;
14 import akka.actor.ActorSystem;
15 import akka.actor.Address;
16 import akka.cluster.Cluster;
17 import akka.cluster.ClusterEvent.CurrentClusterState;
18 import akka.cluster.Member;
19 import akka.cluster.MemberStatus;
20 import akka.cluster.UniqueAddress;
21 import akka.testkit.JavaTestKit;
22 import com.google.common.base.Stopwatch;
23 import com.google.common.collect.Sets;
24 import com.google.common.util.concurrent.Uninterruptibles;
25 import com.typesafe.config.ConfigFactory;
27 import java.net.URISyntaxException;
28 import java.util.ArrayList;
29 import java.util.Arrays;
30 import java.util.Collections;
31 import java.util.List;
33 import java.util.Optional;
35 import java.util.concurrent.TimeUnit;
36 import org.junit.After;
37 import org.junit.AfterClass;
38 import org.junit.Assert;
39 import org.junit.Before;
40 import org.junit.BeforeClass;
41 import org.junit.Test;
42 import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;
43 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
44 import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
45 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
46 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
47 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.UpdateRemoteEndpoints;
48 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.RemoteRpcEndpoint;
49 import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
50 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets;
51 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply;
52 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions;
53 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply;
54 import org.opendaylight.controller.sal.connector.api.RpcRouter;
55 import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
56 import org.opendaylight.yangtools.yang.common.QName;
57 import org.slf4j.Logger;
58 import org.slf4j.LoggerFactory;
59 import scala.concurrent.duration.Duration;
60 import scala.concurrent.duration.FiniteDuration;
62 public class RpcRegistryTest {
63 private static final Logger LOG = LoggerFactory.getLogger(RpcRegistryTest.class);
65 private static ActorSystem node1;
66 private static ActorSystem node2;
67 private static ActorSystem node3;
69 private JavaTestKit invoker1;
70 private JavaTestKit invoker2;
71 private JavaTestKit invoker3;
72 private JavaTestKit registrar1;
73 private JavaTestKit registrar2;
74 private JavaTestKit registrar3;
75 private ActorRef registry1;
76 private ActorRef registry2;
77 private ActorRef registry3;
79 private int routeIdCounter = 1;
82 public static void staticSetup() throws InterruptedException {
83 AkkaConfigurationReader reader = () -> ConfigFactory.load();
85 RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").gossipTickInterval("200ms")
86 .withConfigReader(reader).build();
87 RemoteRpcProviderConfig config2 = new RemoteRpcProviderConfig.Builder("memberB").gossipTickInterval("200ms")
88 .withConfigReader(reader).build();
89 RemoteRpcProviderConfig config3 = new RemoteRpcProviderConfig.Builder("memberC").gossipTickInterval("200ms")
90 .withConfigReader(reader).build();
91 node1 = ActorSystem.create("opendaylight-rpc", config1.get());
92 node2 = ActorSystem.create("opendaylight-rpc", config2.get());
93 node3 = ActorSystem.create("opendaylight-rpc", config3.get());
95 waitForMembersUp(node1, Cluster.get(node2).selfUniqueAddress(), Cluster.get(node3).selfUniqueAddress());
96 waitForMembersUp(node2, Cluster.get(node1).selfUniqueAddress(), Cluster.get(node3).selfUniqueAddress());
99 static void waitForMembersUp(final ActorSystem node, final UniqueAddress... addresses) {
100 Set<UniqueAddress> otherMembersSet = Sets.newHashSet(addresses);
101 Stopwatch sw = Stopwatch.createStarted();
102 while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
103 CurrentClusterState state = Cluster.get(node).state();
104 for (Member m : state.getMembers()) {
105 if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.uniqueAddress())
106 && otherMembersSet.isEmpty()) {
111 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
114 fail("Member(s) " + otherMembersSet + " are not Up");
118 public static void staticTeardown() {
119 JavaTestKit.shutdownActorSystem(node1);
120 JavaTestKit.shutdownActorSystem(node2);
121 JavaTestKit.shutdownActorSystem(node3);
125 public void setup() {
126 invoker1 = new JavaTestKit(node1);
127 registrar1 = new JavaTestKit(node1);
128 registry1 = node1.actorOf(RpcRegistry.props(config(node1), invoker1.getRef(), registrar1.getRef()));
129 invoker2 = new JavaTestKit(node2);
130 registrar2 = new JavaTestKit(node2);
131 registry2 = node2.actorOf(RpcRegistry.props(config(node2), invoker2.getRef(), registrar2.getRef()));
132 invoker3 = new JavaTestKit(node3);
133 registrar3 = new JavaTestKit(node3);
134 registry3 = node3.actorOf(RpcRegistry.props(config(node3), invoker3.getRef(), registrar3.getRef()));
137 private static RemoteRpcProviderConfig config(final ActorSystem node) {
138 return new RemoteRpcProviderConfig(node.settings().config());
142 public void teardown() {
143 if (registry1 != null) {
144 node1.stop(registry1);
146 if (registry2 != null) {
147 node2.stop(registry2);
149 if (registry3 != null) {
150 node3.stop(registry3);
153 if (invoker1 != null) {
154 node1.stop(invoker1.getRef());
156 if (invoker2 != null) {
157 node2.stop(invoker2.getRef());
159 if (invoker3 != null) {
160 node3.stop(invoker3.getRef());
163 if (registrar1 != null) {
164 node1.stop(registrar1.getRef());
166 if (registrar2 != null) {
167 node2.stop(registrar2.getRef());
169 if (registrar3 != null) {
170 node3.stop(registrar3.getRef());
175 * One node cluster. 1. Register rpc, ensure router can be found 2. Then remove rpc, ensure its
179 public void testAddRemoveRpcOnSameNode() throws Exception {
180 LOG.info("testAddRemoveRpcOnSameNode starting");
182 Address nodeAddress = node1.provider().getDefaultAddress();
186 List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
188 registry1.tell(new AddOrUpdateRoutes(addedRouteIds), ActorRef.noSender());
190 // Bucket store should get an update bucket message. Updated bucket contains added rpc.
191 final JavaTestKit testKit = new JavaTestKit(node1);
193 Map<Address, Bucket<RoutingTable>> buckets = retrieveBuckets(registry1, testKit, nodeAddress);
194 verifyBucket(buckets.get(nodeAddress), addedRouteIds);
196 Map<Address, Long> versions = retrieveVersions(registry1, testKit);
197 Assert.assertEquals("Version for bucket " + nodeAddress, (Long) buckets.get(nodeAddress).getVersion(),
198 versions.get(nodeAddress));
201 registry1.tell(new RemoveRoutes(addedRouteIds), ActorRef.noSender());
203 // Bucket store should get an update bucket message. Rpc is removed in the updated bucket
205 verifyEmptyBucket(testKit, registry1, nodeAddress);
207 LOG.info("testAddRemoveRpcOnSameNode ending");
212 * Three node cluster. 1. Register rpc on 1 node, ensure 2nd node gets updated 2. Remove rpc on
213 * 1 node, ensure 2nd node gets updated
216 public void testRpcAddRemoveInCluster() throws Exception {
218 LOG.info("testRpcAddRemoveInCluster starting");
220 List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
222 Address node1Address = node1.provider().getDefaultAddress();
225 registry1.tell(new AddOrUpdateRoutes(addedRouteIds), ActorRef.noSender());
227 // Bucket store on node2 should get a message to update its local copy of remote buckets
228 final JavaTestKit testKit = new JavaTestKit(node2);
230 Map<Address, Bucket<RoutingTable>> buckets = retrieveBuckets(registry2, testKit, node1Address);
231 verifyBucket(buckets.get(node1Address), addedRouteIds);
234 registry1.tell(new RemoveRoutes(addedRouteIds), ActorRef.noSender());
236 // Bucket store on node2 should get a message to update its local copy of remote buckets.
237 // Wait for the bucket for node1 to be empty.
239 verifyEmptyBucket(testKit, registry2, node1Address);
241 LOG.info("testRpcAddRemoveInCluster ending");
244 private void verifyEmptyBucket(final JavaTestKit testKit, final ActorRef registry, final Address address)
245 throws AssertionError {
246 Map<Address, Bucket<RoutingTable>> buckets;
249 buckets = retrieveBuckets(registry1, testKit, address);
252 verifyBucket(buckets.get(address), Collections.<RouteIdentifier<?, ?, ?>>emptyList());
254 } catch (AssertionError e) {
255 if (++numTries >= 50) {
260 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
265 * Three node cluster. Register rpc on 2 nodes. Ensure 3rd gets updated.
268 public void testRpcAddedOnMultiNodes() throws Exception {
269 final JavaTestKit testKit = new JavaTestKit(node3);
272 List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds1 = createRouteIds();
273 registry1.tell(new AddOrUpdateRoutes(addedRouteIds1), ActorRef.noSender());
275 final UpdateRemoteEndpoints req1 = registrar3.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
276 UpdateRemoteEndpoints.class);
279 List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds2 = createRouteIds();
280 registry2.tell(new AddOrUpdateRoutes(addedRouteIds2), ActorRef.noSender());
282 final UpdateRemoteEndpoints req2 = registrar3.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
283 UpdateRemoteEndpoints.class);
284 Address node2Address = node2.provider().getDefaultAddress();
285 Address node1Address = node1.provider().getDefaultAddress();
287 Map<Address, Bucket<RoutingTable>> buckets = retrieveBuckets(registry3, testKit, node1Address,
290 verifyBucket(buckets.get(node1Address), addedRouteIds1);
291 verifyBucket(buckets.get(node2Address), addedRouteIds2);
293 Map<Address, Long> versions = retrieveVersions(registry3, testKit);
294 Assert.assertEquals("Version for bucket " + node1Address, (Long) buckets.get(node1Address).getVersion(),
295 versions.get(node1Address));
296 Assert.assertEquals("Version for bucket " + node2Address, (Long) buckets.get(node2Address).getVersion(),
297 versions.get(node2Address));
299 assertEndpoints(req1, node1Address, invoker1);
300 assertEndpoints(req2, node2Address, invoker2);
304 private static void assertEndpoints(final UpdateRemoteEndpoints msg, final Address address,
305 final JavaTestKit invoker) {
306 final Map<Address, Optional<RemoteRpcEndpoint>> endpoints = msg.getEndpoints();
307 Assert.assertEquals(1, endpoints.size());
309 final Optional<RemoteRpcEndpoint> maybeEndpoint = endpoints.get(address);
310 Assert.assertNotNull(maybeEndpoint);
311 Assert.assertTrue(maybeEndpoint.isPresent());
313 final RemoteRpcEndpoint endpoint = maybeEndpoint.get();
314 final ActorRef router = endpoint.getRouter();
315 Assert.assertNotNull(router);
317 router.tell("hello", ActorRef.noSender());
318 final String s = invoker.expectMsgClass(Duration.create(3, TimeUnit.SECONDS), String.class);
319 Assert.assertEquals("hello", s);
322 private static Map<Address, Long> retrieveVersions(final ActorRef bucketStore, final JavaTestKit testKit) {
323 bucketStore.tell(new GetBucketVersions(), testKit.getRef());
324 GetBucketVersionsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
325 GetBucketVersionsReply.class);
326 return reply.getVersions();
329 private static void verifyBucket(final Bucket<RoutingTable> bucket,
330 final List<RouteIdentifier<?, ?, ?>> expRouteIds) {
331 RoutingTable table = bucket.getData();
332 Assert.assertNotNull("Bucket RoutingTable is null", table);
333 for (RouteIdentifier<?, ?, ?> r : expRouteIds) {
334 if (!table.contains(r)) {
335 Assert.fail("RoutingTable does not contain " + r + ". Actual: " + table);
339 Assert.assertEquals("RoutingTable size", expRouteIds.size(), table.size());
342 private static Map<Address, Bucket<RoutingTable>> retrieveBuckets(final ActorRef bucketStore,
343 final JavaTestKit testKit, final Address... addresses) {
346 bucketStore.tell(new GetAllBuckets(), testKit.getRef());
347 @SuppressWarnings("unchecked")
348 GetAllBucketsReply<RoutingTable> reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
349 GetAllBucketsReply.class);
351 Map<Address, Bucket<RoutingTable>> buckets = reply.getBuckets();
352 boolean foundAll = true;
353 for (Address addr : addresses) {
354 Bucket<RoutingTable> bucket = buckets.get(addr);
355 if (bucket == null) {
365 if (++numTries >= 50) {
366 Assert.fail("Missing expected buckets for addresses: " + Arrays.toString(addresses)
367 + ", Actual: " + buckets);
370 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
375 public void testAddRoutesConcurrency() throws Exception {
376 final JavaTestKit testKit = new JavaTestKit(node1);
378 final int nRoutes = 500;
379 final RouteIdentifier<?, ?, ?>[] added = new RouteIdentifier<?, ?, ?>[nRoutes];
380 for (int i = 0; i < nRoutes; i++) {
381 final RouteIdentifierImpl routeId = new RouteIdentifierImpl(null,
382 new QName(new URI("/mockrpc"), "type" + i), null);
385 //Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
386 registry1.tell(new AddOrUpdateRoutes(Arrays.<RouteIdentifier<?, ?, ?>>asList(routeId)),
387 ActorRef.noSender());
390 GetAllBuckets getAllBuckets = new GetAllBuckets();
391 FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS);
394 registry1.tell(getAllBuckets, testKit.getRef());
395 @SuppressWarnings("unchecked")
396 GetAllBucketsReply<RoutingTable> reply = testKit.expectMsgClass(duration, GetAllBucketsReply.class);
398 Bucket<RoutingTable> localBucket = reply.getBuckets().values().iterator().next();
399 RoutingTable table = localBucket.getData();
400 if (table != null && table.size() == nRoutes) {
401 for (RouteIdentifier<?, ?, ?> r : added) {
402 Assert.assertEquals("RoutingTable contains " + r, true, table.contains(r));
408 if (++numTries >= 50) {
409 Assert.fail("Expected # routes: " + nRoutes + ", Actual: " + table.size());
412 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
416 private List<RpcRouter.RouteIdentifier<?, ?, ?>> createRouteIds() throws URISyntaxException {
417 QName type = new QName(new URI("/mockrpc"), "mockrpc" + routeIdCounter++);
418 List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new ArrayList<>(1);
419 routeIds.add(new RouteIdentifierImpl(null, type, null));