opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/utils/ActorContextTest.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.cluster.datastore.utils;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertFalse;
  13 import static org.junit.Assert.assertNotEquals;
  14 import static org.junit.Assert.assertNotNull;
  15 import static org.junit.Assert.assertNull;
  16 import static org.junit.Assert.assertSame;
  17 import static org.junit.Assert.assertTrue;
  18 import static org.junit.Assert.fail;
  19 import static org.mockito.Mockito.doReturn;
  20 import static org.mockito.Mockito.mock;
  21
  22 import akka.actor.ActorRef;
  23 import akka.actor.ActorSelection;
  24 import akka.actor.ActorSystem;
  25 import akka.actor.Address;
  26 import akka.actor.Props;
  27 import akka.actor.UntypedAbstractActor;
  28 import akka.dispatch.Futures;
  29 import akka.japi.Creator;
  30 import akka.testkit.TestActorRef;
  31 import akka.testkit.javadsl.TestKit;
  32 import akka.util.Timeout;
  33 import com.google.common.base.Optional;
  34 import com.google.common.collect.Maps;
  35 import com.google.common.collect.Sets;
  36 import com.typesafe.config.ConfigFactory;
  37 import java.time.Duration;
  38 import java.util.Arrays;
  39 import java.util.Map;
  40 import java.util.concurrent.TimeUnit;
  41 import org.junit.Assert;
  42 import org.junit.Test;
  43 import org.mockito.Mockito;
  44 import org.opendaylight.controller.cluster.datastore.AbstractActorTest;
  45 import org.opendaylight.controller.cluster.datastore.ClusterWrapper;
  46 import org.opendaylight.controller.cluster.datastore.DataStoreVersions;
  47 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  48 import org.opendaylight.controller.cluster.datastore.DatastoreContextFactory;
  49 import org.opendaylight.controller.cluster.datastore.config.Configuration;
  50 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
  51 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
  52 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
  53 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  54 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
  55 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
  56 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  57 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
  58 import org.opendaylight.controller.cluster.datastore.messages.PrimaryShardInfo;
  59 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
  60 import org.opendaylight.controller.cluster.raft.utils.EchoActor;
  61 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  62 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
  64 import org.slf4j.Logger;
  65 import org.slf4j.LoggerFactory;
  66 import scala.concurrent.Await;
  67 import scala.concurrent.Future;
  68 import scala.concurrent.duration.FiniteDuration;
  69
  70 public class ActorContextTest extends AbstractActorTest {
  71
  72     static final Logger LOG = LoggerFactory.getLogger(ActorContextTest.class);
  73
  74     private static class TestMessage {
  75     }
  76
  77     private static final class MockShardManager extends UntypedAbstractActor {
  78
  79         private final boolean found;
  80         private final ActorRef actorRef;
  81         private final Map<String,Object> findPrimaryResponses = Maps.newHashMap();
  82
  83         private MockShardManager(final boolean found, final ActorRef actorRef) {
  84
  85             this.found = found;
  86             this.actorRef = actorRef;
  87         }
  88
  89         @Override public void onReceive(final Object message) {
  90             if (message instanceof FindPrimary) {
  91                 FindPrimary fp = (FindPrimary)message;
  92                 Object resp = findPrimaryResponses.get(fp.getShardName());
  93                 if (resp == null) {
  94                     LOG.error("No expected FindPrimary response found for shard name {}", fp.getShardName());
  95                 } else {
  96                     getSender().tell(resp, getSelf());
  97                 }
  98
  99                 return;
 100             }
 101
 102             if (found) {
 103                 getSender().tell(new LocalShardFound(actorRef), getSelf());
 104             } else {
 105                 getSender().tell(new LocalShardNotFound(((FindLocalShard) message).getShardName()), getSelf());
 106             }
 107         }
 108
 109         void addFindPrimaryResp(final String shardName, final Object resp) {
 110             findPrimaryResponses.put(shardName, resp);
 111         }
 112
 113         private static Props props(final boolean found, final ActorRef actorRef) {
 114             return Props.create(new MockShardManagerCreator(found, actorRef));
 115         }
 116
 117         private static Props props() {
 118             return Props.create(new MockShardManagerCreator());
 119         }
 120
 121         @SuppressWarnings("serial")
 122         private static class MockShardManagerCreator implements Creator<MockShardManager> {
 123             final boolean found;
 124             final ActorRef actorRef;
 125
 126             MockShardManagerCreator() {
 127                 this.found = false;
 128                 this.actorRef = null;
 129             }
 130
 131             MockShardManagerCreator(final boolean found, final ActorRef actorRef) {
 132                 this.found = found;
 133                 this.actorRef = actorRef;
 134             }
 135
 136             @Override
 137             public MockShardManager create() {
 138                 return new MockShardManager(found, actorRef);
 139             }
 140         }
 141     }
 142
 143     @Test
 144     public void testFindLocalShardWithShardFound() {
 145         final TestKit testKit = new TestKit(getSystem());
 146         testKit.within(Duration.ofSeconds(1), () -> {
 147             ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 148
 149             ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
 150
 151             ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
 152                 mock(ClusterWrapper.class), mock(Configuration.class));
 153
 154             Optional<ActorRef> out = actorContext.findLocalShard("default");
 155
 156             assertEquals(shardActorRef, out.get());
 157
 158             testKit.expectNoMessage();
 159             return null;
 160         });
 161     }
 162
 163     @Test
 164     public void testFindLocalShardWithShardNotFound() {
 165         ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(false, null));
 166
 167         ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef, mock(ClusterWrapper.class),
 168             mock(Configuration.class));
 169
 170         Optional<ActorRef> out = actorContext.findLocalShard("default");
 171         assertFalse(out.isPresent());
 172     }
 173
 174     @Test
 175     public void testExecuteRemoteOperation() {
 176         ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 177
 178         ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
 179
 180         ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
 181             mock(ClusterWrapper.class), mock(Configuration.class));
 182
 183         ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 184
 185         Object out = actorContext.executeOperation(actor, "hello");
 186
 187         assertEquals("hello", out);
 188     }
 189
 190     @Test
 191     public void testExecuteRemoteOperationAsync() throws Exception {
 192         ActorRef shardActorRef = getSystem().actorOf(Props.create(EchoActor.class));
 193
 194         ActorRef shardManagerActorRef = getSystem().actorOf(MockShardManager.props(true, shardActorRef));
 195
 196         ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
 197             mock(ClusterWrapper.class), mock(Configuration.class));
 198
 199         ActorSelection actor = actorContext.actorSelection(shardActorRef.path());
 200
 201         Future<Object> future = actorContext.executeOperationAsync(actor, "hello");
 202
 203         Object result = Await.result(future, FiniteDuration.create(3, TimeUnit.SECONDS));
 204         assertEquals("Result", "hello", result);
 205     }
 206
 207     @Test
 208     public void testIsPathLocal() {
 209         MockClusterWrapper clusterWrapper = new MockClusterWrapper();
 210         ActorContext actorContext = null;
 211
 212         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 213         assertFalse(actorContext.isPathLocal(null));
 214         assertFalse(actorContext.isPathLocal(""));
 215
 216         clusterWrapper.setSelfAddress(null);
 217         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 218         assertFalse(actorContext.isPathLocal(""));
 219
 220         // even if the path is in local format, match the primary path (first 3 elements) and return true
 221         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 222         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 223         assertTrue(actorContext.isPathLocal("akka://test/user/$a"));
 224
 225         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 226         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 227         assertTrue(actorContext.isPathLocal("akka://test/user/$a"));
 228
 229         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 230         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 231         assertTrue(actorContext.isPathLocal("akka://test/user/token2/token3/$a"));
 232
 233         // self address of remote format,but Tx path local format.
 234         clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
 235         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 236         assertTrue(actorContext.isPathLocal("akka://system/user/shardmanager/shard/transaction"));
 237
 238         // self address of local format,but Tx path remote format.
 239         clusterWrapper.setSelfAddress(new Address("akka", "system"));
 240         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 241         assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2550/user/shardmanager/shard/transaction"));
 242
 243         //local path but not same
 244         clusterWrapper.setSelfAddress(new Address("akka", "test"));
 245         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 246         assertTrue(actorContext.isPathLocal("akka://test1/user/$a"));
 247
 248         //ip and port same
 249         clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
 250         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 251         assertTrue(actorContext.isPathLocal("akka://system@127.0.0.1:2550/"));
 252
 253         // forward-slash missing in address
 254         clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
 255         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 256         assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2550"));
 257
 258         //ips differ
 259         clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
 260         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 261         assertFalse(actorContext.isPathLocal("akka://system@127.1.0.1:2550/"));
 262
 263         //ports differ
 264         clusterWrapper.setSelfAddress(new Address("akka", "system", "127.0.0.1", 2550));
 265         actorContext = new ActorContext(getSystem(), null, clusterWrapper, mock(Configuration.class));
 266         assertFalse(actorContext.isPathLocal("akka://system@127.0.0.1:2551/"));
 267     }
 268
 269     @Test
 270     public void testClientDispatcherIsGlobalDispatcher() {
 271         ActorContext actorContext = new ActorContext(getSystem(), mock(ActorRef.class), mock(ClusterWrapper.class),
 272                 mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
 273
 274         assertEquals(getSystem().dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
 275     }
 276
 277     @Test
 278     public void testClientDispatcherIsNotGlobalDispatcher() {
 279         ActorSystem actorSystem = ActorSystem.create("with-custom-dispatchers",
 280                 ConfigFactory.load("application-with-custom-dispatchers.conf"));
 281
 282         ActorContext actorContext = new ActorContext(actorSystem, mock(ActorRef.class), mock(ClusterWrapper.class),
 283                 mock(Configuration.class), DatastoreContext.newBuilder().build(), new PrimaryShardInfoFutureCache());
 284
 285         assertNotEquals(actorSystem.dispatchers().defaultGlobalDispatcher(), actorContext.getClientDispatcher());
 286
 287         actorSystem.terminate();
 288     }
 289
 290     @Test
 291     public void testSetDatastoreContext() {
 292         final TestKit testKit = new TestKit(getSystem());
 293         ActorContext actorContext = new ActorContext(getSystem(), testKit.getRef(),
 294             mock(ClusterWrapper.class), mock(Configuration.class), DatastoreContext.newBuilder()
 295             .operationTimeoutInSeconds(5).shardTransactionCommitTimeoutInSeconds(7).build(),
 296             new PrimaryShardInfoFutureCache());
 297
 298         assertEquals("getOperationDuration", 5, actorContext.getOperationDuration().toSeconds());
 299         assertEquals("getTransactionCommitOperationTimeout", 7,
 300             actorContext.getTransactionCommitOperationTimeout().duration().toSeconds());
 301
 302         DatastoreContext newContext = DatastoreContext.newBuilder().operationTimeoutInSeconds(6)
 303                 .shardTransactionCommitTimeoutInSeconds(8).build();
 304
 305         DatastoreContextFactory mockContextFactory = mock(DatastoreContextFactory.class);
 306         Mockito.doReturn(newContext).when(mockContextFactory).getBaseDatastoreContext();
 307
 308         actorContext.setDatastoreContext(mockContextFactory);
 309
 310         testKit.expectMsgClass(Duration.ofSeconds(5), DatastoreContextFactory.class);
 311
 312         Assert.assertSame("getDatastoreContext", newContext, actorContext.getDatastoreContext());
 313
 314         assertEquals("getOperationDuration", 6, actorContext.getOperationDuration().toSeconds());
 315         assertEquals("getTransactionCommitOperationTimeout", 8,
 316             actorContext.getTransactionCommitOperationTimeout().duration().toSeconds());
 317     }
 318
 319     @Test
 320     public void testFindPrimaryShardAsyncRemotePrimaryFound() throws Exception {
 321
 322         ActorRef shardManager = getSystem().actorOf(MessageCollectorActor.props());
 323
 324         DatastoreContext dataStoreContext = DatastoreContext.newBuilder()
 325                 .logicalStoreType(LogicalDatastoreType.CONFIGURATION)
 326                 .shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
 327
 328         final String expPrimaryPath = "akka://test-system/find-primary-shard";
 329         final short expPrimaryVersion = DataStoreVersions.CURRENT_VERSION;
 330         ActorContext actorContext = new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
 331                 mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
 332             @Override
 333             protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
 334                 return Futures.successful((Object) new RemotePrimaryShardFound(expPrimaryPath, expPrimaryVersion));
 335             }
 336         };
 337
 338         Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
 339         PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
 340
 341         assertNotNull(actual);
 342         assertFalse("LocalShardDataTree present", actual.getLocalShardDataTree().isPresent());
 343         assertTrue("Unexpected PrimaryShardActor path " + actual.getPrimaryShardActor().path(),
 344                 expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
 345         assertEquals("getPrimaryShardVersion", expPrimaryVersion, actual.getPrimaryShardVersion());
 346
 347         Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
 348
 349         PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
 350
 351         assertEquals(cachedInfo, actual);
 352
 353         actorContext.getPrimaryShardInfoCache().remove("foobar");
 354
 355         cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
 356
 357         assertNull(cached);
 358     }
 359
 360     @Test
 361     public void testFindPrimaryShardAsyncLocalPrimaryFound() throws Exception {
 362
 363         ActorRef shardManager = getSystem().actorOf(MessageCollectorActor.props());
 364
 365         DatastoreContext dataStoreContext = DatastoreContext.newBuilder()
 366                 .logicalStoreType(LogicalDatastoreType.CONFIGURATION)
 367                 .shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
 368
 369         final DataTree mockDataTree = Mockito.mock(DataTree.class);
 370         final String expPrimaryPath = "akka://test-system/find-primary-shard";
 371         ActorContext actorContext = new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
 372                 mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
 373             @Override
 374             protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
 375                 return Futures.successful((Object) new LocalPrimaryShardFound(expPrimaryPath, mockDataTree));
 376             }
 377         };
 378
 379         Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
 380         PrimaryShardInfo actual = Await.result(foobar, FiniteDuration.apply(5000, TimeUnit.MILLISECONDS));
 381
 382         assertNotNull(actual);
 383         assertTrue("LocalShardDataTree present", actual.getLocalShardDataTree().isPresent());
 384         assertSame("LocalShardDataTree", mockDataTree, actual.getLocalShardDataTree().get());
 385         assertTrue("Unexpected PrimaryShardActor path " + actual.getPrimaryShardActor().path(),
 386                 expPrimaryPath.endsWith(actual.getPrimaryShardActor().pathString()));
 387         assertEquals("getPrimaryShardVersion", DataStoreVersions.CURRENT_VERSION, actual.getPrimaryShardVersion());
 388
 389         Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
 390
 391         PrimaryShardInfo cachedInfo = Await.result(cached, FiniteDuration.apply(1, TimeUnit.MILLISECONDS));
 392
 393         assertEquals(cachedInfo, actual);
 394
 395         actorContext.getPrimaryShardInfoCache().remove("foobar");
 396
 397         cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
 398
 399         assertNull(cached);
 400     }
 401
 402     @Test
 403     public void testFindPrimaryShardAsyncPrimaryNotFound() {
 404         testFindPrimaryExceptions(new PrimaryNotFoundException("not found"));
 405     }
 406
 407     @Test
 408     public void testFindPrimaryShardAsyncActorNotInitialized() {
 409         testFindPrimaryExceptions(new NotInitializedException("not initialized"));
 410     }
 411
 412     @SuppressWarnings("checkstyle:IllegalCatch")
 413     private static void testFindPrimaryExceptions(final Object expectedException) {
 414         ActorRef shardManager = getSystem().actorOf(MessageCollectorActor.props());
 415
 416         DatastoreContext dataStoreContext = DatastoreContext.newBuilder()
 417                 .logicalStoreType(LogicalDatastoreType.CONFIGURATION)
 418                 .shardLeaderElectionTimeout(100, TimeUnit.MILLISECONDS).build();
 419
 420         ActorContext actorContext =
 421             new ActorContext(getSystem(), shardManager, mock(ClusterWrapper.class),
 422                 mock(Configuration.class), dataStoreContext, new PrimaryShardInfoFutureCache()) {
 423                 @Override
 424                 protected Future<Object> doAsk(final ActorRef actorRef, final Object message, final Timeout timeout) {
 425                     return Futures.successful(expectedException);
 426                 }
 427             };
 428
 429         Future<PrimaryShardInfo> foobar = actorContext.findPrimaryShardAsync("foobar");
 430
 431         try {
 432             Await.result(foobar, FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
 433             fail("Expected" + expectedException.getClass().toString());
 434         } catch (Exception e) {
 435             if (!expectedException.getClass().isInstance(e)) {
 436                 fail("Expected Exception of type " + expectedException.getClass().toString());
 437             }
 438         }
 439
 440         Future<PrimaryShardInfo> cached = actorContext.getPrimaryShardInfoCache().getIfPresent("foobar");
 441
 442         assertNull(cached);
 443     }
 444
 445     @Test
 446     public void testBroadcast() {
 447         ActorRef shardActorRef1 = getSystem().actorOf(MessageCollectorActor.props());
 448         ActorRef shardActorRef2 = getSystem().actorOf(MessageCollectorActor.props());
 449
 450         TestActorRef<MockShardManager> shardManagerActorRef = TestActorRef.create(getSystem(),
 451             MockShardManager.props());
 452         MockShardManager shardManagerActor = shardManagerActorRef.underlyingActor();
 453         shardManagerActor.addFindPrimaryResp("shard1", new RemotePrimaryShardFound(
 454             shardActorRef1.path().toString(), DataStoreVersions.CURRENT_VERSION));
 455         shardManagerActor.addFindPrimaryResp("shard2", new RemotePrimaryShardFound(
 456             shardActorRef2.path().toString(), DataStoreVersions.CURRENT_VERSION));
 457         shardManagerActor.addFindPrimaryResp("shard3", new NoShardLeaderException("not found"));
 458
 459         Configuration mockConfig = mock(Configuration.class);
 460         doReturn(Sets.newLinkedHashSet(Arrays.asList("shard1", "shard2", "shard3"))).when(mockConfig)
 461         .getAllShardNames();
 462
 463         ActorContext actorContext = new ActorContext(getSystem(), shardManagerActorRef,
 464             mock(ClusterWrapper.class), mockConfig,
 465             DatastoreContext.newBuilder().shardInitializationTimeout(200, TimeUnit.MILLISECONDS).build(),
 466             new PrimaryShardInfoFutureCache());
 467
 468         actorContext.broadcast(v -> new TestMessage(), TestMessage.class);
 469
 470         MessageCollectorActor.expectFirstMatching(shardActorRef1, TestMessage.class);
 471         MessageCollectorActor.expectFirstMatching(shardActorRef2, TestMessage.class);
 472     }
 473 }