opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeRemotingTest.java

   1 /*
   2  * Copyright (c) 2016 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.sharding;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.junit.Assert.fail;
  15 import static org.mockito.Mockito.doReturn;
  16 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.findLocalShard;
  17 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.waitUntilShardIsDown;
  18
  19 import akka.actor.ActorRef;
  20 import akka.actor.ActorSystem;
  21 import akka.actor.Address;
  22 import akka.actor.AddressFromURIString;
  23 import akka.cluster.Cluster;
  24 import akka.testkit.JavaTestKit;
  25 import com.google.common.collect.Lists;
  26 import com.typesafe.config.ConfigFactory;
  27 import java.util.Collections;
  28 import java.util.HashSet;
  29 import java.util.Set;
  30 import org.junit.After;
  31 import org.junit.Assert;
  32 import org.junit.Before;
  33 import org.junit.Ignore;
  34 import org.junit.Test;
  35 import org.mockito.Mockito;
  36 import org.opendaylight.controller.cluster.ActorSystemProvider;
  37 import org.opendaylight.controller.cluster.datastore.AbstractTest;
  38 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  39 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  40 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  41 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
  42 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  43 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  44 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  45 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
  46 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  47 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  48 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  49 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
  50 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  51 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
  52 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
  53 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  56 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  57 import org.slf4j.Logger;
  58 import org.slf4j.LoggerFactory;
  59
  60 public class DistributedShardedDOMDataTreeRemotingTest extends AbstractTest {
  61
  62     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class);
  63
  64     private static final Address MEMBER_1_ADDRESS =
  65             AddressFromURIString.parse("akka://cluster-test@127.0.0.1:2558");
  66
  67     private static final DOMDataTreeIdentifier TEST_ID =
  68             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
  69
  70     private static final String MODULE_SHARDS_CONFIG = "module-shards-default.conf";
  71
  72     private ActorSystem leaderSystem;
  73     private ActorSystem followerSystem;
  74
  75
  76     private final Builder leaderDatastoreContextBuilder =
  77             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
  78
  79     private final DatastoreContext.Builder followerDatastoreContextBuilder =
  80             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
  81
  82     private DistributedDataStore leaderConfigDatastore;
  83     private DistributedDataStore leaderOperDatastore;
  84
  85     private DistributedDataStore followerConfigDatastore;
  86     private DistributedDataStore followerOperDatastore;
  87
  88
  89     private IntegrationTestKit followerTestKit;
  90     private IntegrationTestKit leaderTestKit;
  91     private DistributedShardedDOMDataTree leaderShardFactory;
  92
  93     private DistributedShardedDOMDataTree followerShardFactory;
  94     private ActorSystemProvider leaderSystemProvider;
  95     private ActorSystemProvider followerSystemProvider;
  96
  97     @Before
  98     public void setUp() {
  99         InMemoryJournal.clear();
 100         InMemorySnapshotStore.clear();
 101
 102         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 103         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 104
 105         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
 106         Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
 107
 108         leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
 109         doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();
 110
 111         followerSystemProvider = Mockito.mock(ActorSystemProvider.class);
 112         doReturn(followerSystem).when(followerSystemProvider).getActorSystem();
 113
 114     }
 115
 116     @After
 117     public void tearDown() {
 118         if (leaderConfigDatastore != null) {
 119             leaderConfigDatastore.close();
 120         }
 121         if (leaderOperDatastore != null) {
 122             leaderOperDatastore.close();
 123         }
 124
 125         if (followerConfigDatastore != null) {
 126             followerConfigDatastore.close();
 127         }
 128         if (followerOperDatastore != null) {
 129             followerOperDatastore.close();
 130         }
 131
 132         JavaTestKit.shutdownActorSystem(leaderSystem, null, Boolean.TRUE);
 133         JavaTestKit.shutdownActorSystem(followerSystem, null, Boolean.TRUE);
 134
 135         InMemoryJournal.clear();
 136         InMemorySnapshotStore.clear();
 137     }
 138
 139     private void initEmptyDatastores() throws Exception {
 140         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 141
 142         leaderConfigDatastore = leaderTestKit.setupDistributedDataStore(
 143                 "config", MODULE_SHARDS_CONFIG, true,
 144                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 145         leaderOperDatastore = leaderTestKit.setupDistributedDataStore(
 146                 "operational", MODULE_SHARDS_CONFIG, true,
 147                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 148
 149         leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystemProvider,
 150                 leaderOperDatastore,
 151                 leaderConfigDatastore);
 152
 153         followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
 154
 155         followerConfigDatastore = followerTestKit.setupDistributedDataStore(
 156                 "config", MODULE_SHARDS_CONFIG, true, SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 157         followerOperDatastore = followerTestKit.setupDistributedDataStore(
 158                 "operational", MODULE_SHARDS_CONFIG, true,
 159                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
 160
 161         followerShardFactory = new DistributedShardedDOMDataTree(followerSystemProvider,
 162                 followerOperDatastore,
 163                 followerConfigDatastore);
 164
 165         followerTestKit.waitForMembersUp("member-1");
 166
 167         leaderShardFactory.init();
 168         followerShardFactory.init();
 169
 170         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 171                 ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
 172
 173         leaderTestKit.waitUntilLeader(leaderOperDatastore.getActorContext(),
 174                 ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
 175     }
 176
 177     @Test
 178     public void testProducerRegistrations() throws Exception {
 179         LOG.info("testProducerRegistrations starting");
 180         initEmptyDatastores();
 181
 182         leaderTestKit.waitForMembersUp("member-2");
 183
 184         // TODO refactor shard creation and verification to own method
 185         final DistributedShardRegistration shardRegistration =
 186                 waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 187                         TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 188                         DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 189
 190         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 191                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 192
 193         final ActorRef leaderShardManager = leaderConfigDatastore.getActorContext().getShardManager();
 194
 195         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 196                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
 197
 198         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 199                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
 200
 201         final Set<String> peers  = new HashSet<>();
 202         IntegrationTestKit.verifyShardState(leaderConfigDatastore,
 203                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
 204                         peers.addAll(onDemandShardState.getPeerAddresses().values()));
 205         assertEquals(peers.size(), 1);
 206
 207         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
 208         try {
 209             followerShardFactory.createProducer(Collections.singleton(TEST_ID));
 210             fail("Producer should be already registered on the other node");
 211         } catch (final IllegalArgumentException e) {
 212             assertTrue(e.getMessage().contains("is attached to producer"));
 213         }
 214
 215         producer.close();
 216
 217         final DOMDataTreeProducer followerProducer =
 218                 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
 219         try {
 220             leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
 221             fail("Producer should be already registered on the other node");
 222         } catch (final IllegalArgumentException e) {
 223             assertTrue(e.getMessage().contains("is attached to producer"));
 224         }
 225
 226         followerProducer.close();
 227         // try to create a shard on an already registered prefix on follower
 228         try {
 229             waitOnAsyncTask(followerShardFactory.createDistributedShard(
 230                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 231                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 232             fail("This prefix already should have a shard registration that was forwarded from the other node");
 233         } catch (final DOMDataTreeShardingConflictException e) {
 234             assertTrue(e.getMessage().contains("is already occupied by another shard"));
 235         }
 236
 237         shardRegistration.close().toCompletableFuture().get();
 238
 239         LOG.info("testProducerRegistrations ending");
 240     }
 241
 242     @Test
 243     public void testWriteIntoMultipleShards() throws Exception {
 244         LOG.info("testWriteIntoMultipleShards starting");
 245         initEmptyDatastores();
 246
 247         leaderTestKit.waitForMembersUp("member-2");
 248
 249         LOG.debug("registering first shard");
 250         final DistributedShardRegistration shardRegistration =
 251                 waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 252                         TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 253                         DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 254
 255
 256         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 257                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 258         findLocalShard(followerConfigDatastore.getActorContext(),
 259                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 260
 261         final Set<String> peers  = new HashSet<>();
 262         IntegrationTestKit.verifyShardState(leaderConfigDatastore,
 263                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
 264                         peers.addAll(onDemandShardState.getPeerAddresses().values()));
 265         assertEquals(peers.size(), 1);
 266
 267         LOG.debug("Got after waiting for nonleader");
 268         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
 269
 270         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
 271         final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
 272         Assert.assertNotNull(cursor);
 273         final YangInstanceIdentifier nameId =
 274                 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
 275         cursor.write(nameId.getLastPathArgument(),
 276                 ImmutableLeafNodeBuilder.<String>create().withNodeIdentifier(
 277                         new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build());
 278
 279         cursor.close();
 280         LOG.warn("Got to pre submit");
 281
 282         tx.submit().checkedGet();
 283
 284         shardRegistration.close().toCompletableFuture().get();
 285
 286         LOG.info("testWriteIntoMultipleShards ending");
 287     }
 288
 289     @Test
 290     public void testMultipleShardRegistrations() throws Exception {
 291         LOG.info("testMultipleShardRegistrations starting");
 292         initEmptyDatastores();
 293
 294         final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 295                 TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 296                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 297
 298         final DistributedShardRegistration reg2 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 299                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_CONTAINER_PATH),
 300                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 301                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 302
 303         final DistributedShardRegistration reg3 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 304                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.INNER_LIST_PATH),
 305                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 306                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 307
 308         final DistributedShardRegistration reg4 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 309                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.JUNK_PATH),
 310                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 311                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 312
 313         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 314                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 315         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 316                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
 317         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 318                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
 319         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 320                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
 321
 322         // check leader has local shards
 323         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 324                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 325
 326         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 327                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
 328
 329         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 330                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
 331
 332         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 333                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
 334
 335         // check follower has local shards
 336         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 337                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 338
 339         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 340                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
 341
 342         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 343                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
 344
 345         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 346                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
 347
 348         LOG.debug("Closing registrations");
 349
 350         reg1.close().toCompletableFuture().get();
 351         reg2.close().toCompletableFuture().get();
 352         reg3.close().toCompletableFuture().get();
 353         reg4.close().toCompletableFuture().get();
 354
 355         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
 356                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 357
 358         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
 359                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
 360
 361         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
 362                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
 363
 364         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
 365                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
 366
 367         LOG.debug("All leader shards gone");
 368
 369         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
 370                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 371
 372         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
 373                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
 374
 375         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
 376                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
 377
 378         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
 379                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
 380
 381         LOG.debug("All follower shards gone");
 382         LOG.info("testMultipleShardRegistrations ending");
 383     }
 384
 385     @Test
 386     @Ignore
 387     public void testMultipleRegistrationsAtOnePrefix() throws Exception {
 388         LOG.info("testMultipleRegistrationsAtOnePrefix starting");
 389         initEmptyDatastores();
 390
 391         for (int i = 0; i < 10; i++) {
 392             LOG.debug("Round {}", i);
 393             final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
 394                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
 395                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 396
 397             leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
 398                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 399
 400             assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
 401                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 402
 403             assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
 404                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 405
 406
 407             final Set<String> peers  = new HashSet<>();
 408             IntegrationTestKit.verifyShardState(leaderConfigDatastore,
 409                     ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
 410                             peers.addAll(onDemandShardState.getPeerAddresses().values()));
 411             assertEquals(peers.size(), 1);
 412
 413             waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
 414
 415             waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
 416                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 417
 418             waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
 419                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 420         }
 421
 422         LOG.info("testMultipleRegistrationsAtOnePrefix ending");
 423     }
 424 }