X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FIntegrationTestKit.java;h=9570e0e4aa7d5e36823e6ae3379b5cf6800097fe;hb=5c08efbf3a3d046a7650f1b32a6cacfb1e4a6d0d;hp=356026019f7a19ca54a1cd956231b01f9fe6e947;hpb=2f77e92af7a68b4a97dbfb709c6cc9b11a49878a;p=controller.git 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 356026019f..9570e0e4aa 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 @@ -22,18 +22,23 @@ import com.google.common.base.Stopwatch; import com.google.common.collect.Sets; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.Uninterruptibles; +import java.lang.reflect.Constructor; 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.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; @@ -41,12 +46,16 @@ 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; @@ -59,36 +68,73 @@ public class IntegrationTestKit extends ShardTestKit { return datastoreContextBuilder; } - public AbstractDataStore setupDistributedDataStore(final String typeName, final String... shardNames) { - return setupDistributedDataStore(typeName, "module-shards.conf", true, SchemaContextHelper.full(), shardNames); + public DistributedDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, + final boolean waitUntilLeader, + final SchemaContext schemaContext) throws Exception { + return setupDistributedDataStore(typeName, moduleShardsConfig, "modules.conf", waitUntilLeader, schemaContext); + } + + public DistributedDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, + final String modulesConfig, + final boolean waitUntilLeader, + final SchemaContext schemaContext, + final String... shardNames) throws Exception { + return (DistributedDataStore) setupAbstractDataStore(DistributedDataStore.class, typeName, moduleShardsConfig, + modulesConfig, waitUntilLeader, schemaContext, shardNames); + } + + public AbstractDataStore setupAbstractDataStore(final Class implementation, + final String typeName, final String... shardNames) + throws Exception { + return setupAbstractDataStore(implementation, typeName, "module-shards.conf", true, + SchemaContextHelper.full(), shardNames); } - public AbstractDataStore setupDistributedDataStore(final String typeName, final boolean waitUntilLeader, - final String... shardNames) { - return setupDistributedDataStore(typeName, "module-shards.conf", waitUntilLeader, + public AbstractDataStore setupAbstractDataStore(final Class implementation, + final String typeName, final boolean waitUntilLeader, + final String... shardNames) throws Exception { + return setupAbstractDataStore(implementation, typeName, "module-shards.conf", waitUntilLeader, SchemaContextHelper.full(), shardNames); } - public AbstractDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, - final boolean waitUntilLeader, final String... shardNames) { - return setupDistributedDataStore(typeName, moduleShardsConfig, waitUntilLeader, + public AbstractDataStore setupAbstractDataStore(final Class implementation, + final String typeName, final String moduleShardsConfig, + final boolean waitUntilLeader, final String... shardNames) + throws Exception { + return setupAbstractDataStore(implementation, typeName, moduleShardsConfig, waitUntilLeader, SchemaContextHelper.full(), shardNames); } - public AbstractDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig, - final boolean waitUntilLeader, final SchemaContext schemaContext, final String... shardNames) { + public AbstractDataStore setupAbstractDataStore(final Class implementation, + final String typeName, final String moduleShardsConfig, + final boolean waitUntilLeader, + final SchemaContext schemaContext, + final String... shardNames) throws Exception { + return setupAbstractDataStore(implementation, typeName, moduleShardsConfig, "modules.conf", waitUntilLeader, + schemaContext, shardNames); + } + + private AbstractDataStore setupAbstractDataStore(final Class implementation, + final String typeName, final String moduleShardsConfig, + final String modulesConfig, final boolean waitUntilLeader, + final SchemaContext schemaContext, final String... shardNames) + throws Exception { final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem()); - final Configuration config = new ConfigurationImpl(moduleShardsConfig, "modules.conf"); + final Configuration config = new ConfigurationImpl(moduleShardsConfig, modulesConfig); datastoreContextBuilder.dataStoreName(typeName); - DatastoreContext datastoreContext = datastoreContextBuilder.build(); - DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + final DatastoreContext datastoreContext = datastoreContextBuilder.build(); + final DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext(); Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); - AbstractDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, mockContextFactory, - restoreFromSnapshot); + final Constructor constructor = implementation.getDeclaredConstructor( + ActorSystem.class, ClusterWrapper.class, Configuration.class, + DatastoreContextFactory.class, DatastoreSnapshot.class); + + final AbstractDataStore dataStore = (AbstractDataStore) constructor.newInstance( + getSystem(), cluster, config, mockContextFactory, restoreFromSnapshot); dataStore.onGlobalContextUpdated(schemaContext); @@ -122,6 +168,30 @@ public class IntegrationTestKit extends ShardTestKit { 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); @@ -171,6 +241,19 @@ public class IntegrationTestKit extends ShardTestKit { return shard; } + public static void waitUntilShardIsDown(final ActorContext actorContext, final 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(); @@ -196,6 +279,31 @@ public class IntegrationTestKit extends ShardTestKit { throw lastError; } + 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 {