X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FDistributedDataStoreRemotingIntegrationTest.java;h=8100bdc6a2ec7331baa31f26a02682674a60e3ad;hp=be8ede9d6f4a893dc3c2851bc83557dfb4867f44;hb=fd38fb36d0992a3a2391837f6e12615ed98a8bb5;hpb=103ceecd0195cca6c87fbd62a687d8addf128784 diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java index be8ede9d6f..8100bdc6a2 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreRemotingIntegrationTest.java @@ -9,19 +9,36 @@ package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.actor.Address; import akka.actor.AddressFromURIString; import akka.cluster.Cluster; +import akka.pattern.AskTimeoutException; import akka.testkit.JavaTestKit; import com.google.common.base.Optional; +import com.google.common.collect.ImmutableMap; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.Uninterruptibles; import com.typesafe.config.ConfigFactory; import java.math.BigInteger; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.mockito.invocation.InvocationOnMock; +import org.mockito.stubbing.Answer; +import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; +import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException; +import org.opendaylight.controller.cluster.datastore.exceptions.ShardLeaderNotRespondingException; import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction; import org.opendaylight.controller.cluster.datastore.modification.MergeModification; @@ -31,19 +48,29 @@ import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.md.cluster.datastore.model.CarsModel; import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; +import org.opendaylight.controller.md.cluster.datastore.model.TestModel; +import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; +import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener; +import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; +import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction; +import org.opendaylight.controller.md.sal.dom.api.DOMTransactionChain; +import org.opendaylight.controller.sal.core.spi.data.DOMStore; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; +import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode; import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode; import org.opendaylight.yangtools.yang.data.api.schema.MapNode; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification; import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree; +import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType; import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder; +import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder; import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; /** @@ -55,14 +82,21 @@ public class DistributedDataStoreRemotingIntegrationTest { private static final String[] SHARD_NAMES = {"cars", "people"}; + private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"); + private static final Address MEMBER_2_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2559"); + + private static final String MODULE_SHARDS_CONFIG_2 = "module-shards-member1-and-2.conf"; + private static final String MODULE_SHARDS_CONFIG_3 = "module-shards-member1-and-2-and-3.conf"; + private ActorSystem leaderSystem; private ActorSystem followerSystem; + private ActorSystem follower2System; private final DatastoreContext.Builder leaderDatastoreContextBuilder = DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1); private final DatastoreContext.Builder followerDatastoreContextBuilder = - DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(200); + DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); private DistributedDataStore followerDistributedDataStore; private DistributedDataStore leaderDistributedDataStore; @@ -70,35 +104,40 @@ public class DistributedDataStoreRemotingIntegrationTest { private IntegrationTestKit leaderTestKit; @Before - public void setUpClass() { + public void setUp() { leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1")); - Address member1Address = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558"); - Cluster.get(leaderSystem).join(member1Address); + Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS); followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2")); - Cluster.get(followerSystem).join(member1Address); + Cluster.get(followerSystem).join(MEMBER_1_ADDRESS); + + follower2System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member3")); + Cluster.get(follower2System).join(MEMBER_1_ADDRESS); } @After - public void tearDownClass() { + public void tearDown() { JavaTestKit.shutdownActorSystem(leaderSystem); JavaTestKit.shutdownActorSystem(followerSystem); + JavaTestKit.shutdownActorSystem(follower2System); } private void initDatastores(String type) { + initDatastores(type, MODULE_SHARDS_CONFIG_2); + } + + private void initDatastores(String type, String moduleShardsConfig) { leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder); - String moduleShardsConfig = "module-shards-member1-and-2.conf"; + leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, SHARD_NAMES); followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder); followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, SHARD_NAMES); - leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, true, SHARD_NAMES); - leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), SHARD_NAMES); } - private void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception { + private static void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception { Optional> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); @@ -110,14 +149,14 @@ public class DistributedDataStoreRemotingIntegrationTest { assertEquals("Car list node", listBuilder.build(), optional.get()); } - private void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode expNode) + private static void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode expNode) throws Exception { Optional> optional = readTx.read(path).get(5, TimeUnit.SECONDS); assertEquals("isPresent", true, optional.isPresent()); assertEquals("Data node", expNode, optional.get()); } - private void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception { + private static void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception { Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS); assertEquals("exists", true, exists); } @@ -355,14 +394,136 @@ public class DistributedDataStoreRemotingIntegrationTest { assertEquals("isPresent", false, optional.isPresent()); } + @Test + public void testChainedTransactionFailureWithSingleShard() throws Exception { + initDatastores("testChainedTransactionFailureWithSingleShard"); + + ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker( + ImmutableMap.builder().put( + LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(), + MoreExecutors.directExecutor()); + + TransactionChainListener listener = Mockito.mock(TransactionChainListener.class); + DOMTransactionChain txChain = broker.createTransactionChain(listener); + + DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction(); + + ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier( + new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)). + withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); + + writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData); + + try { + writeTx.submit().checkedGet(5, TimeUnit.SECONDS); + fail("Expected TransactionCommitFailedException"); + } catch (TransactionCommitFailedException e) { + // Expected + } + + verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx), any(Throwable.class)); + + txChain.close(); + broker.close(); + } + + @Test + public void testChainedTransactionFailureWithMultipleShards() throws Exception { + initDatastores("testChainedTransactionFailureWithMultipleShards"); + + ConcurrentDOMDataBroker broker = new ConcurrentDOMDataBroker( + ImmutableMap.builder().put( + LogicalDatastoreType.CONFIGURATION, followerDistributedDataStore).build(), + MoreExecutors.directExecutor()); + + TransactionChainListener listener = Mockito.mock(TransactionChainListener.class); + DOMTransactionChain txChain = broker.createTransactionChain(listener); + + DOMDataWriteTransaction writeTx = txChain.newWriteOnlyTransaction(); + + writeTx.put(LogicalDatastoreType.CONFIGURATION, PeopleModel.BASE_PATH, PeopleModel.emptyContainer()); + + ContainerNode invalidData = ImmutableContainerNodeBuilder.create().withNodeIdentifier( + new YangInstanceIdentifier.NodeIdentifier(CarsModel.BASE_QNAME)). + withChild(ImmutableNodes.leafNode(TestModel.JUNK_QNAME, "junk")).build(); + + // Note that merge will validate the data and fail but put succeeds b/c deep validation is not + // done for put for performance reasons. + writeTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.BASE_PATH, invalidData); + + try { + writeTx.submit().checkedGet(5, TimeUnit.SECONDS); + fail("Expected TransactionCommitFailedException"); + } catch (TransactionCommitFailedException e) { + // Expected + } + + verify(listener, timeout(5000)).onTransactionChainFailed(eq(txChain), eq(writeTx), any(Throwable.class)); + + txChain.close(); + broker.close(); + } + + @Test + public void testSingleShardTransactionsWithLeaderChanges() throws Exception { + String testName = "testSingleShardTransactionsWithLeaderChanges"; + initDatastores(testName); + + String followerCarShardName = "member-2-shard-cars-" + testName; + InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 1, ApplyJournalEntries.class ); + + // Write top-level car container from the follower so it uses a remote Tx. + + DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction(); + + writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()); + + followerTestKit.doCommit(writeTx.ready()); + + InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName); + + // Switch the leader to the follower + + followerDatastoreContextBuilder.shardElectionTimeoutFactor(1); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); + + JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + + followerTestKit.waitUntilNoLeader(followerDistributedDataStore.getActorContext(), SHARD_NAMES); + + leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1")); + Cluster.get(leaderSystem).join(MEMBER_2_ADDRESS); + + DatastoreContext.Builder newMember1Builder = DatastoreContext.newBuilder(). + shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); + IntegrationTestKit newMember1TestKit = new IntegrationTestKit(leaderSystem, newMember1Builder); + newMember1TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG_2, false, SHARD_NAMES); + + followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), SHARD_NAMES); + + // Write a car entry to the new leader - should switch to local Tx + + writeTx = followerDistributedDataStore.newWriteOnlyTransaction(); + + MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima"); + writeTx.merge(car1Path, car1); + + followerTestKit.doCommit(writeTx.ready()); + + verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1); + } + @Test public void testReadyLocalTransactionForwardedToLeader() throws Exception { initDatastores("testReadyLocalTransactionForwardedToLeader"); + followerTestKit.waitUntilLeader(followerDistributedDataStore.getActorContext(), "cars"); Optional carsFollowerShard = followerDistributedDataStore.getActorContext().findLocalShard("cars"); assertEquals("Cars follower shard found", true, carsFollowerShard.isPresent()); - TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create(); + TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create(TreeType.OPERATIONAL); dataTree.setSchemaContext(SchemaContextHelper.full()); DataTreeModification modification = dataTree.takeSnapshot().newModification(); @@ -376,8 +537,138 @@ public class DistributedDataStoreRemotingIntegrationTest { ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(transactionID , modification, true); carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef()); - followerTestKit.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS); + Object resp = followerTestKit.expectMsgClass(Object.class); + if(resp instanceof akka.actor.Status.Failure) { + throw new AssertionError("Unexpected failure response", ((akka.actor.Status.Failure)resp).cause()); + } + + assertTrue("Expected response of type " + CommitTransactionReply.SERIALIZABLE_CLASS, + CommitTransactionReply.SERIALIZABLE_CLASS.equals(resp.getClass())); verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car); } + + @Test(expected=NoShardLeaderException.class) + public void testTransactionWithIsolatedLeader() throws Throwable { + leaderDatastoreContextBuilder.shardIsolatedLeaderCheckIntervalInMillis(300); + String testName = "testTransactionWithIsolatedLeader"; + initDatastores(testName); + + JavaTestKit.shutdownActorSystem(followerSystem, null, true); + + Uninterruptibles.sleepUninterruptibly(leaderDistributedDataStore.getActorContext().getDatastoreContext() + .getShardRaftConfig().getElectionTimeOutInterval().toMillis() * 3, TimeUnit.MILLISECONDS); + + DOMStoreWriteTransaction writeTx = leaderDistributedDataStore.newWriteOnlyTransaction(); + writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + + try { + followerTestKit.doCommit(writeTx.ready()); + } catch (ExecutionException e) { + throw e.getCause(); + } + } + + @Test(expected=AskTimeoutException.class) + public void testTransactionWithShardLeaderNotResponding() throws Throwable { + followerDatastoreContextBuilder.shardElectionTimeoutFactor(30); + initDatastores("testTransactionWithShardLeaderNotResponding"); + + // Do an initial read to get the primary shard info cached. + + DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction(); + readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS); + + // Shutdown the leader and try to create a new tx. + + JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + + followerDatastoreContextBuilder.operationTimeoutInMillis(50).shardElectionTimeoutFactor(1); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); + + DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); + + rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + + try { + followerTestKit.doCommit(rwTx.ready()); + } catch (ExecutionException e) { + assertTrue("Expected ShardLeaderNotRespondingException cause. Actual: " + e.getCause(), + e.getCause() instanceof ShardLeaderNotRespondingException); + assertNotNull("Expected a nested cause", e.getCause().getCause()); + throw e.getCause().getCause(); + } + } + + @Test(expected=NoShardLeaderException.class) + public void testTransactionWithCreateTxFailureDueToNoLeader() throws Throwable { + initDatastores("testTransactionWithCreateTxFailureDueToNoLeader"); + + // Do an initial read to get the primary shard info cached. + + DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction(); + readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS); + + // Shutdown the leader and try to create a new tx. + + JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + + Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); + + followerDatastoreContextBuilder.operationTimeoutInMillis(10).shardElectionTimeoutFactor(1); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); + + DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); + + rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + + try { + followerTestKit.doCommit(rwTx.ready()); + } catch (ExecutionException e) { + throw e.getCause(); + } + } + + @Test + public void testTransactionRetryWithInitialAskTimeoutExOnCreateTx() throws Exception { + followerDatastoreContextBuilder.shardElectionTimeoutFactor(30); + String testName = "testTransactionRetryWithInitialAskTimeoutExOnCreateTx"; + initDatastores(testName, MODULE_SHARDS_CONFIG_3); + + DatastoreContext.Builder follower2DatastoreContextBuilder = DatastoreContext.newBuilder(). + shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5); + IntegrationTestKit follower2TestKit = new IntegrationTestKit(follower2System, follower2DatastoreContextBuilder); + follower2TestKit.setupDistributedDataStore(testName, MODULE_SHARDS_CONFIG_3, false, SHARD_NAMES); + + // Do an initial read to get the primary shard info cached. + + DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction(); + readTx.read(CarsModel.BASE_PATH).checkedGet(5, TimeUnit.SECONDS); + + // Shutdown the leader and try to create a new tx. + + JavaTestKit.shutdownActorSystem(leaderSystem, null, true); + + followerDatastoreContextBuilder.operationTimeoutInMillis(500); + sendDatastoreContextUpdate(followerDistributedDataStore, followerDatastoreContextBuilder); + + DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction(); + + rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()); + + followerTestKit.doCommit(rwTx.ready()); + } + + private static void sendDatastoreContextUpdate(DistributedDataStore dataStore, final Builder builder) { + DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + Answer answer = new Answer() { + @Override + public DatastoreContext answer(InvocationOnMock invocation) { + return builder.build(); + } + }; + Mockito.doAnswer(answer).when(mockContextFactory).getBaseDatastoreContext(); + Mockito.doAnswer(answer).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); + dataStore.onDatastoreContextUpdated(mockContextFactory); + } }