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%2FAbstractDistributedDataStoreIntegrationTest.java;h=11f0e3d02a33834528f5947cee4a058603bacfd3;hp=c267a19e1337ac5110dd2300ff076d36208d9191;hb=9503012683559849cd2f01b882b57fdcbcca59f0;hpb=66e553f2098ea61426c4a441be57395673535c2f diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractDistributedDataStoreIntegrationTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractDistributedDataStoreIntegrationTest.java index c267a19e13..11f0e3d02a 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractDistributedDataStoreIntegrationTest.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/AbstractDistributedDataStoreIntegrationTest.java @@ -7,40 +7,37 @@ */ package org.opendaylight.controller.cluster.datastore; +import static org.awaitility.Awaitility.await; +import static org.hamcrest.CoreMatchers.instanceOf; +import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.junit.Assume.assumeFalse; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.timeout; import static org.mockito.Mockito.verify; -import akka.actor.ActorRef; import akka.actor.ActorSystem; import com.google.common.base.Throwables; import com.google.common.collect.ImmutableMap; -import com.google.common.collect.Range; -import com.google.common.primitives.UnsignedLong; import com.google.common.util.concurrent.FluentFuture; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.Uninterruptibles; -import java.math.BigInteger; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; -import java.util.Iterator; import java.util.List; import java.util.Optional; -import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; -import org.junit.Ignore; import org.junit.Test; import org.junit.runners.Parameterized.Parameter; import org.mockito.Mockito; @@ -51,12 +48,13 @@ import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderExc import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard; import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound; import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; -import org.opendaylight.controller.cluster.datastore.persisted.FrontendHistoryMetadata; +import org.opendaylight.controller.cluster.datastore.persisted.FrontendClientMetadata; import org.opendaylight.controller.cluster.datastore.persisted.FrontendShardDataTreeSnapshotMetadata; import org.opendaylight.controller.cluster.datastore.persisted.MetadataShardDataTreeSnapshot; import org.opendaylight.controller.cluster.datastore.persisted.ShardSnapshotState; import org.opendaylight.controller.cluster.datastore.utils.MockDataTreeChangeListener; import org.opendaylight.controller.cluster.raft.persisted.Snapshot; +import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore; 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; @@ -74,6 +72,7 @@ import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort; import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain; import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction; import org.opendaylight.yangtools.concepts.ListenerRegistration; +import org.opendaylight.yangtools.yang.common.Uint64; 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; @@ -84,6 +83,7 @@ import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeConfiguratio import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes; import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder; import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory; +import org.opendaylight.yangtools.yang.model.api.SchemaContext; public abstract class AbstractDistributedDataStoreIntegrationTest { @@ -138,7 +138,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { writeTx = dataStore.newWriteOnlyTransaction(); - final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000)); final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima"); writeTx.write(carPath, car); @@ -151,7 +151,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { // Verify the data in the store final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction(); - Optional> optional = readTx.read(carPath).get(5, TimeUnit.SECONDS); + Optional optional = readTx.read(carPath).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", car, optional.get()); @@ -173,14 +173,14 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { // 2. Write some data final YangInstanceIdentifier nodePath = TestModel.TEST_PATH; - final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + final NormalizedNode nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME); readWriteTx.write(nodePath, nodeToWrite); // 3. Read the data from Tx final Boolean exists = readWriteTx.exists(nodePath).get(5, TimeUnit.SECONDS); assertEquals("exists", Boolean.TRUE, exists); - Optional> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS); + Optional optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", nodeToWrite, optional.get()); @@ -222,7 +222,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { readWriteTx = dataStore.newReadWriteTransaction(); - final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000)); final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima"); readWriteTx.write(carPath, car); @@ -233,7 +233,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final Boolean exists = readWriteTx.exists(carPath).get(5, TimeUnit.SECONDS); assertEquals("exists", Boolean.TRUE, exists); - Optional> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS); + Optional optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", car, optional.get()); @@ -253,7 +253,6 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { } @Test - @Ignore("Flushes a closed tx leak in single node, needs to be handled separately") public void testSingleTransactionsWritesInQuickSuccession() throws Exception { final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), datastoreContextBuilder); try (AbstractDataStore dataStore = testKit.setupAbstractDataStore( @@ -269,43 +268,56 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { int numCars = 5; for (int i = 0; i < numCars; i++) { writeTx = txChain.newWriteOnlyTransaction(); - writeTx.write(CarsModel.newCarPath("car" + i), - CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000))); + writeTx.write(CarsModel.newCarPath("car" + i), CarsModel.newCarEntry("car" + i, Uint64.valueOf(20000))); testKit.doCommit(writeTx.ready()); - DOMStoreReadTransaction domStoreReadTransaction = txChain.newReadOnlyTransaction(); - domStoreReadTransaction.read(CarsModel.BASE_PATH).get(); - - domStoreReadTransaction.close(); + try (var tx = txChain.newReadOnlyTransaction()) { + tx.read(CarsModel.BASE_PATH).get(); + } } - // verify frontend metadata has no holes in purged transactions causing overtime memory leak - Optional localShard = dataStore.getActorUtils().findLocalShard("cars-1"); - FrontendShardDataTreeSnapshotMetadata frontendMetadata = - (FrontendShardDataTreeSnapshotMetadata) dataStore.getActorUtils() - .executeOperation(localShard.get(), new RequestFrontendMetadata()); - - if (dataStore.getActorUtils().getDatastoreContext().isUseTellBasedProtocol()) { - Iterator iterator = - frontendMetadata.getClients().get(0).getCurrentHistories().iterator(); - FrontendHistoryMetadata metadata = iterator.next(); - while (iterator.hasNext() && metadata.getHistoryId() != 1) { - metadata = iterator.next(); - } - Set> ranges = metadata.getPurgedTransactions().asRanges(); + // wait to let the shard catch up with purged + await("transaction state propagation").atMost(5, TimeUnit.SECONDS) + .pollInterval(500, TimeUnit.MILLISECONDS) + .untilAsserted(() -> { + // verify frontend metadata has no holes in purged transactions causing overtime memory leak + final var localShard = dataStore.getActorUtils().findLocalShard("cars-1") .orElseThrow(); + FrontendShardDataTreeSnapshotMetadata frontendMetadata = + (FrontendShardDataTreeSnapshotMetadata) dataStore.getActorUtils() + .executeOperation(localShard, new RequestFrontendMetadata()); + + final var clientMeta = frontendMetadata.getClients().get(0); + if (dataStore.getActorUtils().getDatastoreContext().isUseTellBasedProtocol()) { + assertTellMetadata(clientMeta); + } else { + assertAskMetadata(clientMeta); + } + }); - assertEquals(1, ranges.size()); - } else { - // ask based should track no metadata - assertTrue(frontendMetadata.getClients().get(0).getCurrentHistories().isEmpty()); - } + final var body = txChain.newReadOnlyTransaction().read(CarsModel.CAR_LIST_PATH) + .get(5, TimeUnit.SECONDS) + .orElseThrow() + .body(); + assertThat(body, instanceOf(Collection.class)); + assertEquals("# cars", numCars, ((Collection) body).size()); + } + } - final Optional> optional = txChain.newReadOnlyTransaction() - .read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); - assertTrue("isPresent", optional.isPresent()); - assertEquals("# cars", numCars, ((Collection) optional.get().getValue()).size()); + private static void assertAskMetadata(final FrontendClientMetadata clientMeta) { + // FIXME: needs to be enabled + assumeFalse(true); + // ask based should track no metadata + assertEquals(List.of(), clientMeta.getCurrentHistories()); + } + + private static void assertTellMetadata(final FrontendClientMetadata clientMeta) { + final var iterator = clientMeta.getCurrentHistories().iterator(); + var metadata = iterator.next(); + while (iterator.hasNext() && metadata.getHistoryId() != 1) { + metadata = iterator.next(); } + assertEquals("[[0..10]]", metadata.getPurgedTransactions().ranges().toString()); } @SuppressWarnings("checkstyle:IllegalCatch") @@ -442,7 +454,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { assertNotNull("newWriteOnlyTransaction returned null", writeTx); // 2. Write some data - final NormalizedNode testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); + final NormalizedNode testNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME); writeTx.write(TestModel.TEST_PATH, testNode); // 3. Ready the Tx for commit @@ -468,7 +480,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { // the data from the first // Tx is visible after being readied. DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction(); - Optional> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); + Optional optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", testNode, optional.get()); @@ -533,7 +545,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction(); - final MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000)); + final MapEntryNode car = CarsModel.newCarEntry("optima", Uint64.valueOf(20000)); final YangInstanceIdentifier carPath = CarsModel.newCarPath("optima"); readWriteTx.write(carPath, car); @@ -541,7 +553,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack"); readWriteTx.merge(personPath, person); - Optional> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS); + Optional optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", car, optional.get()); @@ -603,7 +615,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final DOMDataTreeReadWriteTransaction rwTx = txChain.newReadWriteTransaction(); rwTx.merge(LogicalDatastoreType.CONFIGURATION, CarsModel.newCarPath("car" + i), - CarsModel.newCarEntry("car" + i, BigInteger.valueOf(20000))); + CarsModel.newCarEntry("car" + i, Uint64.valueOf(20000))); futures.add(rwTx.commit()); } @@ -612,10 +624,10 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { f.get(5, TimeUnit.SECONDS); } - final Optional> optional = txChain.newReadOnlyTransaction() + final Optional optional = txChain.newReadOnlyTransaction() .read(LogicalDatastoreType.CONFIGURATION, CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); - assertEquals("# cars", numCars, ((Collection) optional.get().getValue()).size()); + assertEquals("# cars", numCars, ((Collection) optional.get().body()).size()); txChain.close(); @@ -637,7 +649,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final DOMStoreReadWriteTransaction rwTx2 = txChain.newReadWriteTransaction(); - final Optional> optional = rwTx2.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); + final Optional optional = rwTx2.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS); assertFalse("isPresent", optional.isPresent()); txChain.close(); @@ -693,10 +705,10 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final DOMStoreThreePhaseCommitCohort cohort1 = writeTx.ready(); // Create read-only tx's and issue a read. - FluentFuture>> readFuture1 = txChain + FluentFuture> readFuture1 = txChain .newReadOnlyTransaction().read(TestModel.TEST_PATH); - FluentFuture>> readFuture2 = txChain + FluentFuture> readFuture2 = txChain .newReadOnlyTransaction().read(TestModel.TEST_PATH); // Create another write tx and issue the write. @@ -863,13 +875,13 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final String name = "transactionIntegrationTest"; final ContainerNode carsNode = CarsModel.newCarsNode( - CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", BigInteger.valueOf(20000L)), - CarsModel.newCarEntry("sportage", BigInteger.valueOf(30000L)))); + CarsModel.newCarsMapNode(CarsModel.newCarEntry("optima", Uint64.valueOf(20000)), + CarsModel.newCarEntry("sportage", Uint64.valueOf(30000)))); DataTree dataTree = new InMemoryDataTreeFactory().create( DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full()); AbstractShardTest.writeToStore(dataTree, CarsModel.BASE_PATH, carsNode); - NormalizedNode root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.empty()); + NormalizedNode root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.empty()); final Snapshot carsSnapshot = Snapshot.create( new ShardSnapshotState(new MetadataShardDataTreeSnapshot(root)), @@ -878,7 +890,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { dataTree = new InMemoryDataTreeFactory().create(DataTreeConfiguration.DEFAULT_OPERATIONAL, SchemaContextHelper.full()); - final NormalizedNode peopleNode = PeopleModel.create(); + final NormalizedNode peopleNode = PeopleModel.create(); AbstractShardTest.writeToStore(dataTree, PeopleModel.BASE_PATH, peopleNode); root = AbstractShardTest.readStore(dataTree, YangInstanceIdentifier.empty()); @@ -897,7 +909,7 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { final DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction(); // two reads - Optional> optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS); + Optional optional = readTx.read(CarsModel.BASE_PATH).get(5, TimeUnit.SECONDS); assertTrue("isPresent", optional.isPresent()); assertEquals("Data node", carsNode, optional.get()); @@ -906,4 +918,63 @@ public abstract class AbstractDistributedDataStoreIntegrationTest { assertEquals("Data node", peopleNode, optional.get()); } } + + @Test + public void testSnapshotOnRootOverwrite() throws Exception { + if (!DistributedDataStore.class.isAssignableFrom(testParameter)) { + // FIXME: ClientBackedDatastore does not have stable indexes/term, the snapshot index seems to fluctuate + return; + } + + final IntegrationTestKit testKit = new IntegrationTestKit(getSystem(), + datastoreContextBuilder.snapshotOnRootOverwrite(true)); + try (AbstractDataStore dataStore = testKit.setupAbstractDataStore( + testParameter, "testRootOverwrite", "module-shards-default-cars-member1.conf", + true, "cars", "default")) { + + ContainerNode rootNode = ImmutableContainerNodeBuilder.create() + .withNodeIdentifier(YangInstanceIdentifier.NodeIdentifier.create(SchemaContext.NAME)) + .withChild(CarsModel.create()) + .build(); + + testKit.testWriteTransaction(dataStore, YangInstanceIdentifier.empty(), rootNode); + IntegrationTestKit.verifyShardState(dataStore, "cars", + state -> assertEquals(1, state.getSnapshotIndex())); + + // root has been written expect snapshot at index 0 + verifySnapshot("member-1-shard-cars-testRootOverwrite", 1, 1); + + for (int i = 0; i < 10; i++) { + testKit.testWriteTransaction(dataStore, CarsModel.newCarPath("car " + i), + CarsModel.newCarEntry("car " + i, Uint64.ONE)); + } + + // fake snapshot causes the snapshotIndex to move + IntegrationTestKit.verifyShardState(dataStore, "cars", + state -> assertEquals(10, state.getSnapshotIndex())); + + // however the real snapshot still has not changed and was taken at index 0 + verifySnapshot("member-1-shard-cars-testRootOverwrite", 1, 1); + + // root overwrite so expect a snapshot + testKit.testWriteTransaction(dataStore, YangInstanceIdentifier.empty(), rootNode); + + // this was a real snapshot so everything should be in it(1 + 10 + 1) + IntegrationTestKit.verifyShardState(dataStore, "cars", + state -> assertEquals(12, state.getSnapshotIndex())); + + verifySnapshot("member-1-shard-cars-testRootOverwrite", 12, 1); + } + } + + private static void verifySnapshot(final String persistenceId, final long lastAppliedIndex, + final long lastAppliedTerm) { + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> { + List snap = InMemorySnapshotStore.getSnapshots(persistenceId, Snapshot.class); + assertEquals(1, snap.size()); + assertEquals(lastAppliedIndex, snap.get(0).getLastAppliedIndex()); + assertEquals(lastAppliedTerm, snap.get(0).getLastAppliedTerm()); + } + ); + } }