X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fsharding%2FDistributedShardedDOMDataTreeTest.java;h=88a12fe7d799b0b9670f7059ec6719d179f7cefd;hb=c1336f9b497bc6867536a24f629c3f0b002ccb2f;hp=deb6aa5eb7f450f3f12a028ff309fa0dada184a9;hpb=877c428f2897f6e3b11efd25589a84aa0c660a31;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeTest.java index deb6aa5eb7..88a12fe7d7 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/sharding/DistributedShardedDOMDataTreeTest.java @@ -8,8 +8,9 @@ package org.opendaylight.controller.cluster.sharding; -import static org.junit.Assert.assertTrue; -import static org.junit.Assert.fail; +import static org.junit.Assert.assertNotNull; +import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.findLocalShard; +import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.waitUntilShardIsDown; import akka.actor.ActorRef; import akka.actor.ActorSystem; @@ -18,14 +19,16 @@ import akka.actor.AddressFromURIString; import akka.cluster.Cluster; import akka.testkit.JavaTestKit; import com.google.common.collect.Lists; +import com.google.common.util.concurrent.CheckedFuture; import com.typesafe.config.ConfigFactory; +import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import org.junit.After; import org.junit.Assert; import org.junit.Before; import org.junit.Ignore; import org.junit.Test; -import org.mockito.Mockito; import org.opendaylight.controller.cluster.datastore.AbstractTest; import org.opendaylight.controller.cluster.datastore.DatastoreContext; import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; @@ -35,74 +38,68 @@ import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.FindPrimary; import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; -import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound; import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils; -import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; import org.opendaylight.controller.md.cluster.datastore.model.TestModel; import org.opendaylight.mdsal.common.api.LogicalDatastoreType; +import org.opendaylight.mdsal.common.api.TransactionCommitFailedException; import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction; import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier; import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer; -import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException; import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor; -import org.opendaylight.mdsal.dom.broker.ShardedDOMDataTree; +import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier; +import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates; +import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode; +import org.opendaylight.yangtools.yang.data.api.schema.MapNode; +import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder; +import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +@Ignore("distributed-data is broken needs to be removed") public class DistributedShardedDOMDataTreeTest extends AbstractTest { + private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class); + private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"); private static final DOMDataTreeIdentifier TEST_ID = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH); - private ShardedDOMDataTree shardedDOMDataTree = new ShardedDOMDataTree(); - private ActorSystem leaderSystem; - private ActorSystem followerSystem; - private final Builder leaderDatastoreContextBuilder = - DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2); + DatastoreContext.newBuilder() + .shardHeartbeatIntervalInMillis(100) + .shardElectionTimeoutFactor(2) + .logicalStoreType( + org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION); - private final DatastoreContext.Builder followerDatastoreContextBuilder = - DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5) - .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName()); - - private DistributedDataStore followerDistributedDataStore; private DistributedDataStore leaderDistributedDataStore; - private IntegrationTestKit followerTestKit; private IntegrationTestKit leaderTestKit; private DistributedShardedDOMDataTree leaderShardFactory; - private DistributedShardedDOMDataTree followerShardFactory; @Before public void setUp() { - shardedDOMDataTree = new ShardedDOMDataTree(); leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1")); Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS); - followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2")); - Cluster.get(followerSystem).join(MEMBER_1_ADDRESS); } @After public void tearDown() { - if (followerDistributedDataStore != null) { - leaderDistributedDataStore.close(); - } if (leaderDistributedDataStore != null) { leaderDistributedDataStore.close(); } JavaTestKit.shutdownActorSystem(leaderSystem); - JavaTestKit.shutdownActorSystem(followerSystem); } private void initEmptyDatastore(final String type) { @@ -111,90 +108,44 @@ public class DistributedShardedDOMDataTreeTest extends AbstractTest { leaderDistributedDataStore = leaderTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full()); - followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder); - followerDistributedDataStore = - followerTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full()); - leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystem, - Mockito.mock(DistributedDataStore.class), + leaderDistributedDataStore, leaderDistributedDataStore); - - followerShardFactory = new DistributedShardedDOMDataTree(followerSystem, - Mockito.mock(DistributedDataStore.class), - followerDistributedDataStore); } @Test - public void testProducerRegistrations() throws Exception { + public void testWritesIntoDefaultShard() throws Exception { initEmptyDatastore("config"); - final DistributedShardRegistration shardRegistration = - leaderShardFactory.createDistributedShard(TEST_ID, - Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)); + leaderShardFactory.createDistributedShard(TEST_ID, + Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)); leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), - ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())); + ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY)); - final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager(); + final DOMDataTreeIdentifier configRoot = + new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY); - leaderShardManager.tell( - new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef()); - leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalShardFound.class); - - IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorContext(), - ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())); - - leaderShardManager.tell( - new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef()); - leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalPrimaryShardFound.class); - - final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager(); - followerShardManager.tell( - new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), followerTestKit.getRef()); - followerTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), RemotePrimaryShardFound.class); - - final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID)); - try { - followerShardFactory.createProducer(Collections.singleton(TEST_ID)); - fail("Producer should be already registered on the other node"); - } catch (final IllegalArgumentException e) { - assertTrue(e.getMessage().contains("is attached to producer")); - } - - producer.close(); - - final DOMDataTreeProducer followerProducer = - followerShardFactory.createProducer(Collections.singleton(TEST_ID)); - try { - leaderShardFactory.createProducer(Collections.singleton(TEST_ID)); - fail("Producer should be already registered on the other node"); - } catch (final IllegalArgumentException e) { - assertTrue(e.getMessage().contains("is attached to producer")); - } + final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(configRoot)); - followerProducer.close(); - // try to create a shard on an already registered prefix on follower - try { - followerShardFactory.createDistributedShard(TEST_ID, - Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)); - fail("This prefix already should have a shard registration that was forwarded from the other node"); - } catch (final DOMDataTreeShardingConflictException e) { - assertTrue(e.getMessage().contains("is already occupied by shard")); - } + final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true); + final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID); + Assert.assertNotNull(cursor); } @Test - @Ignore("Needs some other stuff related to 5280") - public void testWriteIntoMultipleShards() throws Exception { + public void testSingleNodeWrites() throws Exception { initEmptyDatastore("config"); - final DistributedShardRegistration shardRegistration = - leaderShardFactory.createDistributedShard( - TEST_ID,Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)); + leaderShardFactory.createDistributedShard(TEST_ID, + Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)); + final DistributedShardRegistration shardRegistration = + leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)); leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())); + LOG.warn("Got after waiting for nonleader"); final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager(); new JavaTestKit(leaderSystem) { @@ -203,12 +154,6 @@ public class DistributedShardedDOMDataTreeTest extends AbstractTest { new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef()); expectMsgClass(duration("5 seconds"), LocalShardFound.class); - final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager(); - - followerShardManager.tell( - new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef()); - expectMsgClass(duration("5 seconds"), LocalShardFound.class); - leaderDistributedDataStore.getActorContext().getShardManager().tell( new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef()); expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class); @@ -223,12 +168,136 @@ public class DistributedShardedDOMDataTreeTest extends AbstractTest { final YangInstanceIdentifier nameId = YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build(); cursor.write(nameId.getLastPathArgument(), - ImmutableLeafNodeBuilder.create() - .withNodeIdentifier(new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build()); + ImmutableLeafNodeBuilder.create().withNodeIdentifier( + new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build()); cursor.close(); - tx.submit(); + LOG.warn("Got to pre submit"); + tx.submit().checkedGet(); + } + + @Test + public void testMultipleWritesIntoSingleMapEntry() throws Exception { + initEmptyDatastore("config"); + + final DistributedShardRegistration shardRegistration = + leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)); + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())); + + LOG.warn("Got after waiting for nonleader"); + final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager(); + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); + + final YangInstanceIdentifier oid1 = TestModel.OUTER_LIST_PATH.node(new NodeIdentifierWithPredicates( + TestModel.OUTER_LIST_QNAME, QName.create(TestModel.OUTER_LIST_QNAME, "id"), 0)); + final DOMDataTreeIdentifier outerListPath = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1); + + final DistributedShardRegistration outerListShardReg = leaderShardFactory.createDistributedShard(outerListPath, + Lists.newArrayList(AbstractTest.MEMBER_NAME)); + + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier())); + + final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer( + Collections.singletonList(outerListPath)); + + final DOMDataTreeCursorAwareTransaction tx = shardProducer.createTransaction(false); + final DOMDataTreeWriteCursor cursor = + tx.createCursor(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1)); + assertNotNull(cursor); + + MapNode innerList = ImmutableMapNodeBuilder + .create() + .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME)) + .build(); + + cursor.write(new NodeIdentifier(TestModel.INNER_LIST_QNAME), innerList); + cursor.close(); + tx.submit().checkedGet(); + + final ArrayList> futures = new ArrayList<>(); + for (int i = 0; i < 1000; i++) { + final Collection innerListMapEntries = createInnerListMapEntries(1000, "run-" + i); + for (final MapEntryNode innerListMapEntry : innerListMapEntries) { + final DOMDataTreeCursorAwareTransaction tx1 = shardProducer.createTransaction(false); + final DOMDataTreeWriteCursor cursor1 = tx1.createCursor( + new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, + oid1.node(new NodeIdentifier(TestModel.INNER_LIST_QNAME)))); + cursor1.write(innerListMapEntry.getIdentifier(), innerListMapEntry); + cursor1.close(); + futures.add(tx1.submit()); + } + } + + futures.get(futures.size() - 1).checkedGet(); + + } + + private static Collection createInnerListMapEntries(final int amount, final String valuePrefix) { + final Collection ret = new ArrayList<>(); + for (int i = 0; i < amount; i++) { + ret.add(ImmutableNodes.mapEntryBuilder() + .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.INNER_LIST_QNAME, + QName.create(TestModel.INNER_LIST_QNAME, "name"), Integer.toString(i))) + .withChild(ImmutableNodes + .leafNode(QName.create(TestModel.INNER_LIST_QNAME, "value"), valuePrefix + "-" + i)) + .build()); + } + + return ret; + } + + @Test + public void testDistributedData() throws Exception { + initEmptyDatastore("config"); + + leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME)); + leaderShardFactory.createDistributedShard( + new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_CONTAINER_PATH), + Lists.newArrayList(AbstractTest.MEMBER_NAME)); + leaderShardFactory.createDistributedShard( + new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.INNER_LIST_PATH), + Lists.newArrayList(AbstractTest.MEMBER_NAME)); + leaderShardFactory.createDistributedShard( + new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.JUNK_PATH), + Lists.newArrayList(AbstractTest.MEMBER_NAME)); + + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)); + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)); + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)); + + } + + @Test + public void testMultipleRegistrationsAtOnePrefix() throws Exception { + initEmptyDatastore("config"); + + for (int i = 0; i < 10; i++) { + LOG.debug("Round {}", i); + final DistributedShardRegistration reg1 = leaderShardFactory + .createDistributedShard(TEST_ID, + Lists.newArrayList(AbstractTest.MEMBER_NAME)); + + leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); + + assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.TEST_PATH))); + + reg1.close(); + + waitUntilShardIsDown(leaderDistributedDataStore.getActorContext(), + ClusterUtils.getCleanShardName(TestModel.TEST_PATH)); + + } } -} +} \ No newline at end of file