opendaylight/md-sal/sal-remoterpc-connector/src/test/java/org/opendaylight/controller/remote/rpc/registry/RpcRegistryTest.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.controller.remote.rpc.registry;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.ActorSystem;
  13 import akka.actor.Address;
  14 import akka.actor.Props;
  15 import akka.japi.Pair;
  16 import akka.testkit.JavaTestKit;
  17 import com.google.common.util.concurrent.Uninterruptibles;
  18 import com.typesafe.config.Config;
  19 import com.typesafe.config.ConfigFactory;
  20 import java.net.URI;
  21 import java.net.URISyntaxException;
  22 import java.util.ArrayList;
  23 import java.util.Arrays;
  24 import java.util.Collections;
  25 import java.util.List;
  26 import java.util.Map;
  27 import java.util.concurrent.TimeUnit;
  28 import org.junit.After;
  29 import org.junit.AfterClass;
  30 import org.junit.Assert;
  31 import org.junit.Before;
  32 import org.junit.BeforeClass;
  33 import org.junit.Test;
  34 import org.opendaylight.controller.cluster.common.actor.AkkaConfigurationReader;
  35 import org.opendaylight.controller.remote.rpc.RemoteRpcProviderConfig;
  36 import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
  37 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.AddOrUpdateRoutes;
  38 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRouters;
  39 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRoutersReply;
  40 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.RemoveRoutes;
  41 import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
  42 import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
  43 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBuckets;
  44 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetAllBucketsReply;
  45 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersions;
  46 import org.opendaylight.controller.remote.rpc.registry.gossip.Messages.BucketStoreMessages.GetBucketVersionsReply;
  47 import org.opendaylight.controller.sal.connector.api.RpcRouter;
  48 import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
  49 import org.opendaylight.yangtools.yang.common.QName;
  50 import scala.concurrent.duration.Duration;
  51 import scala.concurrent.duration.FiniteDuration;
  52
  53 public class RpcRegistryTest {
  54
  55     private static ActorSystem node1;
  56     private static ActorSystem node2;
  57     private static ActorSystem node3;
  58
  59     private ActorRef registry1;
  60     private ActorRef registry2;
  61     private ActorRef registry3;
  62
  63     private int routeIdCounter = 1;
  64
  65     @BeforeClass
  66     public static void staticSetup() throws InterruptedException {
  67         AkkaConfigurationReader reader = new AkkaConfigurationReader() {
  68             @Override
  69             public Config read() {
  70                 return ConfigFactory.load();
  71             }
  72         };
  73
  74         RemoteRpcProviderConfig config1 = new RemoteRpcProviderConfig.Builder("memberA").withConfigReader(reader).build();
  75         RemoteRpcProviderConfig config2 = new RemoteRpcProviderConfig.Builder("memberB").withConfigReader(reader).build();
  76         RemoteRpcProviderConfig config3 = new RemoteRpcProviderConfig.Builder("memberC").withConfigReader(reader).build();
  77         node1 = ActorSystem.create("opendaylight-rpc", config1.get());
  78         node2 = ActorSystem.create("opendaylight-rpc", config2.get());
  79         node3 = ActorSystem.create("opendaylight-rpc", config3.get());
  80     }
  81
  82     @AfterClass
  83     public static void staticTeardown() {
  84         JavaTestKit.shutdownActorSystem(node1);
  85         JavaTestKit.shutdownActorSystem(node2);
  86         JavaTestKit.shutdownActorSystem(node3);
  87     }
  88
  89     @Before
  90     public void setup() {
  91         registry1 = node1.actorOf(Props.create(RpcRegistry.class, config(node1)));
  92         registry2 = node2.actorOf(Props.create(RpcRegistry.class, config(node2)));
  93         registry3 = node3.actorOf(Props.create(RpcRegistry.class, config(node3)));
  94     }
  95
  96     private RemoteRpcProviderConfig config(ActorSystem node){
  97         return new RemoteRpcProviderConfig(node.settings().config());
  98     }
  99
 100     @After
 101     public void teardown() {
 102         if (registry1 != null) {
 103             node1.stop(registry1);
 104         }
 105         if (registry2 != null) {
 106             node2.stop(registry2);
 107         }
 108         if (registry3 != null) {
 109             node3.stop(registry3);
 110         }
 111     }
 112
 113     /**
 114      * One node cluster. 1. Register rpc, ensure router can be found 2. Then remove rpc, ensure its
 115      * deleted
 116      *
 117      * @throws URISyntaxException
 118      * @throws InterruptedException
 119      */
 120     @Test
 121     public void testAddRemoveRpcOnSameNode() throws Exception {
 122
 123         System.out.println("testAddRemoveRpcOnSameNode starting");
 124
 125         final JavaTestKit mockBroker = new JavaTestKit(node1);
 126
 127         Address nodeAddress = node1.provider().getDefaultAddress();
 128
 129         // Add rpc on node 1
 130         registry1.tell(new SetLocalRouter(mockBroker.getRef()), mockBroker.getRef());
 131
 132         List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
 133
 134         registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker.getRef());
 135
 136         // Bucket store should get an update bucket message. Updated bucket contains added rpc.
 137
 138         Map<Address, Bucket> buckets = retrieveBuckets(registry1, mockBroker, nodeAddress);
 139         verifyBucket(buckets.get(nodeAddress), addedRouteIds);
 140
 141         Map<Address, Long> versions = retrieveVersions(registry1, mockBroker);
 142         Assert.assertEquals("Version for bucket " + nodeAddress, buckets.get(nodeAddress).getVersion(),
 143                 versions.get(nodeAddress));
 144
 145         // Now remove rpc
 146         registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker.getRef());
 147
 148         // Bucket store should get an update bucket message. Rpc is removed in the updated bucket
 149
 150         verifyEmptyBucket(mockBroker, registry1, nodeAddress);
 151
 152         System.out.println("testAddRemoveRpcOnSameNode ending");
 153
 154     }
 155
 156     /**
 157      * Three node cluster. 1. Register rpc on 1 node, ensure 2nd node gets updated 2. Remove rpc on
 158      * 1 node, ensure 2nd node gets updated
 159      *
 160      * @throws URISyntaxException
 161      * @throws InterruptedException
 162      */
 163     @Test
 164     public void testRpcAddRemoveInCluster() throws Exception {
 165
 166         System.out.println("testRpcAddRemoveInCluster starting");
 167
 168         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 169         final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 170
 171         List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds = createRouteIds();
 172
 173         Address node1Address = node1.provider().getDefaultAddress();
 174
 175         // Add rpc on node 1
 176         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 177         registry1.tell(new AddOrUpdateRoutes(addedRouteIds), mockBroker1.getRef());
 178
 179         // Bucket store on node2 should get a message to update its local copy of remote buckets
 180
 181         Map<Address, Bucket> buckets = retrieveBuckets(registry2, mockBroker2, node1Address);
 182         verifyBucket(buckets.get(node1Address), addedRouteIds);
 183
 184         // Now remove
 185         registry1.tell(new RemoveRoutes(addedRouteIds), mockBroker1.getRef());
 186
 187         // Bucket store on node2 should get a message to update its local copy of remote buckets.
 188         // Wait for the bucket for node1 to be empty.
 189
 190         verifyEmptyBucket(mockBroker2, registry2, node1Address);
 191
 192         System.out.println("testRpcAddRemoveInCluster ending");
 193     }
 194
 195     private void verifyEmptyBucket(JavaTestKit testKit, ActorRef registry, Address address)
 196             throws AssertionError {
 197         Map<Address, Bucket> buckets;
 198         int nTries = 0;
 199         while(true) {
 200             buckets = retrieveBuckets(registry1, testKit, address);
 201
 202             try {
 203                 verifyBucket(buckets.get(address), Collections.<RouteIdentifier<?, ?, ?>>emptyList());
 204                 break;
 205             } catch (AssertionError e) {
 206                 if(++nTries >= 50) {
 207                     throw e;
 208                 }
 209             }
 210
 211             Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
 212         }
 213     }
 214
 215     /**
 216      * Three node cluster. Register rpc on 2 nodes. Ensure 3rd gets updated.
 217      *
 218      * @throws Exception
 219      */
 220     @Test
 221     public void testRpcAddedOnMultiNodes() throws Exception {
 222
 223         final JavaTestKit mockBroker1 = new JavaTestKit(node1);
 224         final JavaTestKit mockBroker2 = new JavaTestKit(node2);
 225         final JavaTestKit mockBroker3 = new JavaTestKit(node3);
 226
 227         registry3.tell(new SetLocalRouter(mockBroker3.getRef()), mockBroker3.getRef());
 228
 229         // Add rpc on node 1
 230         List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds1 = createRouteIds();
 231         registry1.tell(new SetLocalRouter(mockBroker1.getRef()), mockBroker1.getRef());
 232         registry1.tell(new AddOrUpdateRoutes(addedRouteIds1), mockBroker1.getRef());
 233
 234         // Add rpc on node 2
 235         List<RpcRouter.RouteIdentifier<?, ?, ?>> addedRouteIds2 = createRouteIds();
 236         registry2.tell(new SetLocalRouter(mockBroker2.getRef()), mockBroker2.getRef());
 237         registry2.tell(new AddOrUpdateRoutes(addedRouteIds2), mockBroker2.getRef());
 238
 239         Address node1Address = node1.provider().getDefaultAddress();
 240         Address node2Address = node2.provider().getDefaultAddress();
 241
 242         Map<Address, Bucket> buckets = retrieveBuckets(registry3, mockBroker3, node1Address,
 243                 node2Address);
 244
 245         verifyBucket(buckets.get(node1Address), addedRouteIds1);
 246         verifyBucket(buckets.get(node2Address), addedRouteIds2);
 247
 248         Map<Address, Long> versions = retrieveVersions(registry3, mockBroker3);
 249         Assert.assertEquals("Version for bucket " + node1Address, buckets.get(node1Address).getVersion(),
 250                 versions.get(node1Address));
 251         Assert.assertEquals("Version for bucket " + node2Address, buckets.get(node2Address).getVersion(),
 252                 versions.get(node2Address));
 253
 254         RouteIdentifier<?, ?, ?> routeID = addedRouteIds1.get(0);
 255         registry3.tell(new FindRouters(routeID), mockBroker3.getRef());
 256
 257         FindRoutersReply reply = mockBroker3.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
 258                 FindRoutersReply.class);
 259
 260         List<Pair<ActorRef, Long>> respList = reply.getRouterWithUpdateTime();
 261         Assert.assertEquals("getRouterWithUpdateTime size", 1, respList.size());
 262
 263         respList.get(0).first().tell("hello", ActorRef.noSender());
 264         mockBroker1.expectMsgEquals(Duration.create(3, TimeUnit.SECONDS), "hello");
 265     }
 266
 267     private Map<Address, Long> retrieveVersions(ActorRef bucketStore, JavaTestKit testKit) {
 268         bucketStore.tell(new GetBucketVersions(), testKit.getRef());
 269         GetBucketVersionsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
 270                 GetBucketVersionsReply.class);
 271         return reply.getVersions();
 272     }
 273
 274     private void verifyBucket(Bucket<RoutingTable> bucket, List<RouteIdentifier<?, ?, ?>> expRouteIds) {
 275         RoutingTable table = bucket.getData();
 276         Assert.assertNotNull("Bucket RoutingTable is null", table);
 277         for(RouteIdentifier<?, ?, ?> r: expRouteIds) {
 278             if(!table.contains(r)) {
 279                 Assert.fail("RoutingTable does not contain " + r + ". Actual: " + table);
 280             }
 281         }
 282
 283         Assert.assertEquals("RoutingTable size", expRouteIds.size(), table.size());
 284     }
 285
 286     private Map<Address, Bucket> retrieveBuckets(ActorRef bucketStore, JavaTestKit testKit,
 287             Address... addresses) {
 288         int nTries = 0;
 289         while(true) {
 290             bucketStore.tell(new GetAllBuckets(), testKit.getRef());
 291             GetAllBucketsReply reply = testKit.expectMsgClass(Duration.create(3, TimeUnit.SECONDS),
 292                     GetAllBucketsReply.class);
 293
 294             Map<Address, Bucket> buckets = reply.getBuckets();
 295             boolean foundAll = true;
 296             for(Address addr: addresses) {
 297                 Bucket bucket = buckets.get(addr);
 298                 if(bucket  == null) {
 299                     foundAll = false;
 300                     break;
 301                 }
 302             }
 303
 304             if(foundAll) {
 305                 return buckets;
 306             }
 307
 308             if(++nTries >= 50) {
 309                 Assert.fail("Missing expected buckets for addresses: " + Arrays.toString(addresses)
 310                         + ", Actual: " + buckets);
 311             }
 312
 313             Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
 314         }
 315     }
 316
 317     @SuppressWarnings("unchecked")
 318     @Test
 319     public void testAddRoutesConcurrency() throws Exception {
 320         final JavaTestKit testKit = new JavaTestKit(node1);
 321
 322         registry1.tell(new SetLocalRouter(testKit.getRef()), ActorRef.noSender());
 323
 324         final int nRoutes = 500;
 325         final RouteIdentifier<?, ?, ?>[] added = new RouteIdentifier<?, ?, ?>[nRoutes];
 326         for(int i = 0; i < nRoutes; i++) {
 327             final RouteIdentifierImpl routeId = new RouteIdentifierImpl(null,
 328                     new QName(new URI("/mockrpc"), "type" + i), null);
 329             added[i] = routeId;
 330
 331             //Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 332             registry1.tell(new AddOrUpdateRoutes(Arrays.<RouteIdentifier<?, ?, ?>>asList(routeId)),
 333                     ActorRef.noSender());
 334         }
 335
 336         GetAllBuckets getAllBuckets = new GetAllBuckets();
 337         FiniteDuration duration = Duration.create(3, TimeUnit.SECONDS);
 338         int nTries = 0;
 339         while(true) {
 340             registry1.tell(getAllBuckets, testKit.getRef());
 341             GetAllBucketsReply reply = testKit.expectMsgClass(duration, GetAllBucketsReply.class);
 342
 343             Bucket<RoutingTable> localBucket = reply.getBuckets().values().iterator().next();
 344             RoutingTable table = localBucket.getData();
 345             if(table != null && table.size() == nRoutes) {
 346                 for(RouteIdentifier<?, ?, ?> r: added) {
 347                     Assert.assertEquals("RoutingTable contains " + r, true, table.contains(r));
 348                 }
 349
 350                 break;
 351             }
 352
 353             if(++nTries >= 50) {
 354                 Assert.fail("Expected # routes: " + nRoutes + ", Actual: " + table.size());
 355             }
 356
 357             Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
 358         }
 359     }
 360
 361     private List<RpcRouter.RouteIdentifier<?, ?, ?>> createRouteIds() throws URISyntaxException {
 362         QName type = new QName(new URI("/mockrpc"), "mockrpc" + routeIdCounter++);
 363         List<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new ArrayList<>();
 364         routeIds.add(new RouteIdentifierImpl(null, type, null));
 365         return routeIds;
 366     }
 367 }