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