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%2FIntegrationTestKit.java;h=86c875993ae1e7697c77ff226b5cf6b719dd5e1d;hp=afffa99e0cdfb61c8f96f55d4dd6db0de046ea96;hb=149feb98f151186975fe42bab5853e05aafd4b51;hpb=6f3c16acf17d0cb4d5f44b666751f8db84a652be diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java index afffa99e0c..86c875993a 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/IntegrationTestKit.java @@ -10,6 +10,7 @@ package org.opendaylight.controller.cluster.datastore; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.fail; + import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.cluster.Cluster; @@ -24,14 +25,19 @@ import com.google.common.util.concurrent.Uninterruptibles; import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import org.mockito.Mockito; import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder; import org.opendaylight.controller.cluster.datastore.config.Configuration; import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl; +import org.opendaylight.controller.cluster.datastore.config.EmptyModuleShardConfigProvider; import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; -import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot; +import org.opendaylight.controller.cluster.datastore.messages.OnDemandShardState; +import org.opendaylight.controller.cluster.datastore.persisted.DatastoreSnapshot; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; +import org.opendaylight.controller.cluster.raft.client.messages.GetOnDemandRaftState; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; +import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort; import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain; @@ -39,16 +45,20 @@ 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.NormalizedNode; import org.opendaylight.yangtools.yang.model.api.SchemaContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import scala.concurrent.Await; import scala.concurrent.Future; import scala.concurrent.duration.Duration; public class IntegrationTestKit extends ShardTestKit { + private static final Logger LOG = LoggerFactory.getLogger(IntegrationTestKit.class); + protected DatastoreContext.Builder datastoreContextBuilder; protected DatastoreSnapshot restoreFromSnapshot; - public IntegrationTestKit(ActorSystem actorSystem, Builder datastoreContextBuilder) { + public IntegrationTestKit(final ActorSystem actorSystem, final Builder datastoreContextBuilder) { super(actorSystem); this.datastoreContextBuilder = datastoreContextBuilder; } @@ -57,26 +67,33 @@ public class IntegrationTestKit extends ShardTestKit { return datastoreContextBuilder; } - public DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) { + public AbstractDataStore setupDistributedDataStore(final String typeName, final String... shardNames) { return setupDistributedDataStore(typeName, "module-shards.conf", true, SchemaContextHelper.full(), shardNames); } - public DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader, - String... shardNames) { + public AbstractDataStore setupDistributedDataStore(final String typeName, final boolean waitUntilLeader, + final String... shardNames) { return setupDistributedDataStore(typeName, "module-shards.conf", waitUntilLeader, SchemaContextHelper.full(), shardNames); } - public DistributedDataStore setupDistributedDataStore(String typeName, String moduleShardsConfig, - boolean waitUntilLeader, String... shardNames) { + public AbstractDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, + final boolean waitUntilLeader, final String... shardNames) { return setupDistributedDataStore(typeName, moduleShardsConfig, waitUntilLeader, SchemaContextHelper.full(), shardNames); } - public DistributedDataStore setupDistributedDataStore(String typeName, String moduleShardsConfig, - boolean waitUntilLeader, SchemaContext schemaContext, String... shardNames) { - ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); - Configuration config = new ConfigurationImpl(moduleShardsConfig, "modules.conf"); + public AbstractDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, + final boolean waitUntilLeader, final SchemaContext schemaContext, final String... shardNames) { + return setupDistributedDataStore(typeName, moduleShardsConfig, "modules.conf", waitUntilLeader, + schemaContext, shardNames); + } + + public AbstractDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, + final String modulesConfig, final boolean waitUntilLeader, + final SchemaContext schemaContext, final String... shardNames) { + final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); + final Configuration config = new ConfigurationImpl(moduleShardsConfig, modulesConfig); datastoreContextBuilder.dataStoreName(typeName); @@ -85,12 +102,12 @@ public class IntegrationTestKit extends ShardTestKit { Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext(); Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); - DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, mockContextFactory, + AbstractDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, mockContextFactory, restoreFromSnapshot); dataStore.onGlobalContextUpdated(schemaContext); - if(waitUntilLeader) { + if (waitUntilLeader) { waitUntilLeader(dataStore.getActorContext(), shardNames); } @@ -98,8 +115,54 @@ public class IntegrationTestKit extends ShardTestKit { return dataStore; } - public void waitUntilLeader(ActorContext actorContext, String... shardNames) { - for(String shardName: shardNames) { + public DistributedDataStore setupDistributedDataStoreWithoutConfig(final String typeName, + final SchemaContext schemaContext) { + final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); + final ConfigurationImpl configuration = new ConfigurationImpl(new EmptyModuleShardConfigProvider()); + + getDatastoreContextBuilder().dataStoreName(typeName); + + final DatastoreContext datastoreContext = getDatastoreContextBuilder().build(); + + final DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext(); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); + + final DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, + configuration, mockContextFactory, restoreFromSnapshot); + + dataStore.onGlobalContextUpdated(schemaContext); + + datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext); + return dataStore; + } + + public DistributedDataStore setupDistributedDataStoreWithoutConfig(final String typeName, + final SchemaContext schemaContext, + final LogicalDatastoreType storeType) { + final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); + final ConfigurationImpl configuration = new ConfigurationImpl(new EmptyModuleShardConfigProvider()); + + getDatastoreContextBuilder().dataStoreName(typeName); + + final DatastoreContext datastoreContext = + getDatastoreContextBuilder().logicalStoreType(storeType).build(); + + final DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext(); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); + + final DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, + configuration, mockContextFactory, restoreFromSnapshot); + + dataStore.onGlobalContextUpdated(schemaContext); + + datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext); + return dataStore; + } + + public void waitUntilLeader(final ActorContext actorContext, final String... shardNames) { + for (String shardName: shardNames) { ActorRef shard = findLocalShard(actorContext, shardName); assertNotNull("Shard was not created for " + shardName, shard); @@ -108,8 +171,8 @@ public class IntegrationTestKit extends ShardTestKit { } } - public void waitUntilNoLeader(ActorContext actorContext, String... shardNames) { - for(String shardName: shardNames) { + public void waitUntilNoLeader(final ActorContext actorContext, final String... shardNames) { + for (String shardName: shardNames) { ActorRef shard = findLocalShard(actorContext, shardName); assertNotNull("No local shard found for " + shardName, shard); @@ -117,14 +180,14 @@ public class IntegrationTestKit extends ShardTestKit { } } - public void waitForMembersUp(String... otherMembers) { + public void waitForMembersUp(final String... otherMembers) { Set otherMembersSet = Sets.newHashSet(otherMembers); Stopwatch sw = Stopwatch.createStarted(); - while(sw.elapsed(TimeUnit.SECONDS) <= 10) { + while (sw.elapsed(TimeUnit.SECONDS) <= 10) { CurrentClusterState state = Cluster.get(getSystem()).state(); - for(Member m: state.getMembers()) { - if(m.status() == MemberStatus.up() && otherMembersSet.remove(m.getRoles().iterator().next()) && - otherMembersSet.isEmpty()) { + for (Member m: state.getMembers()) { + if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.getRoles().iterator().next()) + && otherMembersSet.isEmpty()) { return; } } @@ -135,20 +198,33 @@ public class IntegrationTestKit extends ShardTestKit { fail("Member(s) " + otherMembersSet + " are not Up"); } - public static ActorRef findLocalShard(ActorContext actorContext, String shardName) { + public static ActorRef findLocalShard(final ActorContext actorContext, final String shardName) { ActorRef shard = null; - for(int i = 0; i < 20 * 5 && shard == null; i++) { + for (int i = 0; i < 20 * 5 && shard == null; i++) { Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); Optional shardReply = actorContext.findLocalShard(shardName); - if(shardReply.isPresent()) { + if (shardReply.isPresent()) { shard = shardReply.get(); } } return shard; } - public static void verifyShardStats(DistributedDataStore datastore, String shardName, ShardStatsVerifier verifier) - throws Exception { + public static void waitUntilShardIsDown(ActorContext actorContext, String shardName) { + for (int i = 0; i < 20 * 5 ; i++) { + LOG.debug("Waiting for shard down {}", shardName); + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + Optional shardReply = actorContext.findLocalShard(shardName); + if (!shardReply.isPresent()) { + return; + } + } + + throw new IllegalStateException("Shard[" + shardName + " did not shutdown in time"); + } + + public static void verifyShardStats(final AbstractDataStore datastore, final String shardName, + final ShardStatsVerifier verifier) throws Exception { ActorContext actorContext = datastore.getActorContext(); Future future = actorContext.findLocalShardAsync(shardName); @@ -156,9 +232,9 @@ public class IntegrationTestKit extends ShardTestKit { AssertionError lastError = null; Stopwatch sw = Stopwatch.createStarted(); - while(sw.elapsed(TimeUnit.SECONDS) <= 5) { - ShardStats shardStats = (ShardStats)actorContext. - executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE); + while (sw.elapsed(TimeUnit.SECONDS) <= 5) { + ShardStats shardStats = (ShardStats)actorContext + .executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE); try { verifier.verify(shardStats); @@ -172,8 +248,33 @@ public class IntegrationTestKit extends ShardTestKit { throw lastError; } - void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath, - NormalizedNode nodeToWrite) throws Exception { + public static void verifyShardState(final AbstractDataStore datastore, final String shardName, + final Consumer verifier) throws Exception { + ActorContext actorContext = datastore.getActorContext(); + + Future future = actorContext.findLocalShardAsync(shardName); + ActorRef shardActor = Await.result(future, Duration.create(10, TimeUnit.SECONDS)); + + AssertionError lastError = null; + Stopwatch sw = Stopwatch.createStarted(); + while (sw.elapsed(TimeUnit.SECONDS) <= 5) { + OnDemandShardState shardState = (OnDemandShardState)actorContext + .executeOperation(shardActor, GetOnDemandRaftState.INSTANCE); + + try { + verifier.accept(shardState); + return; + } catch (AssertionError e) { + lastError = e; + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + } + } + + throw lastError; + } + + void testWriteTransaction(final AbstractDataStore dataStore, final YangInstanceIdentifier nodePath, + final NormalizedNode nodeToWrite) throws Exception { // 1. Create a write-only Tx @@ -208,47 +309,40 @@ public class IntegrationTestKit extends ShardTestKit { cohort.commit().get(5, TimeUnit.SECONDS); } - void doCommit(final ListenableFuture canCommitFuture, final DOMStoreThreePhaseCommitCohort cohort) throws Exception { + void doCommit(final ListenableFuture canCommitFuture, final DOMStoreThreePhaseCommitCohort cohort) + throws Exception { Boolean canCommit = canCommitFuture.get(7, TimeUnit.SECONDS); assertEquals("canCommit", true, canCommit); cohort.preCommit().get(5, TimeUnit.SECONDS); cohort.commit().get(5, TimeUnit.SECONDS); } - void assertExceptionOnCall(Callable callable, Class expType) + @SuppressWarnings("checkstyle:IllegalCatch") + void assertExceptionOnCall(final Callable callable, final Class expType) throws Exception { try { callable.call(); fail("Expected " + expType.getSimpleName()); - } catch(Exception e) { + } catch (Exception e) { assertEquals("Exception type", expType, e.getClass()); } } void assertExceptionOnTxChainCreates(final DOMStoreTransactionChain txChain, - Class expType) throws Exception { - assertExceptionOnCall(new Callable() { - @Override - public Void call() throws Exception { - txChain.newWriteOnlyTransaction(); - return null; - } + final Class expType) throws Exception { + assertExceptionOnCall(() -> { + txChain.newWriteOnlyTransaction(); + return null; }, expType); - assertExceptionOnCall(new Callable() { - @Override - public Void call() throws Exception { - txChain.newReadWriteTransaction(); - return null; - } + assertExceptionOnCall(() -> { + txChain.newReadWriteTransaction(); + return null; }, expType); - assertExceptionOnCall(new Callable() { - @Override - public Void call() throws Exception { - txChain.newReadOnlyTransaction(); - return null; - } + assertExceptionOnCall(() -> { + txChain.newReadOnlyTransaction(); + return null; }, expType); }