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=afffa99e0cdfb61c8f96f55d4dd6db0de046ea96;hb=4f1f2ae598588f6aa5aac59b2206d97ad402a193;hp=2b2729c9598921cccdbb356b1bfd68c0f9bd87b4;hpb=ffc46de334c8a903844b9f4aff73dc68b2401659;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 2b2729c959..afffa99e0c 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 @@ -12,16 +12,24 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.fail; import akka.actor.ActorRef; import akka.actor.ActorSystem; -import akka.actor.PoisonPill; +import akka.cluster.Cluster; +import akka.cluster.ClusterEvent.CurrentClusterState; +import akka.cluster.Member; +import akka.cluster.MemberStatus; import com.google.common.base.Optional; +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.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.TimeUnit; +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.shardstrategy.ShardStrategyFactory; +import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats; +import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot; import org.opendaylight.controller.cluster.datastore.utils.ActorContext; import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper; import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction; @@ -31,67 +39,103 @@ 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 scala.concurrent.Await; +import scala.concurrent.Future; +import scala.concurrent.duration.Duration; -class IntegrationTestKit extends ShardTestKit { +public class IntegrationTestKit extends ShardTestKit { - DatastoreContext.Builder datastoreContextBuilder; + protected DatastoreContext.Builder datastoreContextBuilder; + protected DatastoreSnapshot restoreFromSnapshot; - IntegrationTestKit(ActorSystem actorSystem, Builder datastoreContextBuilder) { + public IntegrationTestKit(ActorSystem actorSystem, Builder datastoreContextBuilder) { super(actorSystem); this.datastoreContextBuilder = datastoreContextBuilder; } - DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) { - return setupDistributedDataStore(typeName, true, shardNames); + public DatastoreContext.Builder getDatastoreContextBuilder() { + return datastoreContextBuilder; } - DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader, - String... shardNames) { - return setupDistributedDataStore(typeName, "module-shards.conf", waitUntilLeader, shardNames); + public DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) { + return setupDistributedDataStore(typeName, "module-shards.conf", true, SchemaContextHelper.full(), shardNames); } - DistributedDataStore setupDistributedDataStore(String typeName, String moduleShardsConfig, boolean waitUntilLeader, + public DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader, String... shardNames) { + return setupDistributedDataStore(typeName, "module-shards.conf", waitUntilLeader, + SchemaContextHelper.full(), shardNames); + } + + public DistributedDataStore setupDistributedDataStore(String typeName, String moduleShardsConfig, + boolean waitUntilLeader, 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"); - ShardStrategyFactory.setConfiguration(config); - datastoreContextBuilder.dataStoreType(typeName); + datastoreContextBuilder.dataStoreName(typeName); DatastoreContext datastoreContext = datastoreContextBuilder.build(); + DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext(); + Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString()); - DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, datastoreContext); + DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, mockContextFactory, + restoreFromSnapshot); - SchemaContext schemaContext = SchemaContextHelper.full(); dataStore.onGlobalContextUpdated(schemaContext); if(waitUntilLeader) { waitUntilLeader(dataStore.getActorContext(), shardNames); } + datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext); return dataStore; } - void waitUntilLeader(ActorContext actorContext, String... shardNames) { + public void waitUntilLeader(ActorContext actorContext, String... shardNames) { for(String shardName: shardNames) { ActorRef shard = findLocalShard(actorContext, shardName); - assertNotNull("Shard was not created", shard); + assertNotNull("Shard was not created for " + shardName, shard); waitUntilLeader(shard); } } - void waitUntilNoLeader(ActorContext actorContext, String... shardNames) { + public void waitUntilNoLeader(ActorContext actorContext, String... shardNames) { for(String shardName: shardNames) { ActorRef shard = findLocalShard(actorContext, shardName); - assertNotNull("No local shard found", shard); + assertNotNull("No local shard found for " + shardName, shard); waitUntilNoLeader(shard); } } - private ActorRef findLocalShard(ActorContext actorContext, String shardName) { + public void waitForMembersUp(String... otherMembers) { + Set otherMembersSet = Sets.newHashSet(otherMembers); + Stopwatch sw = Stopwatch.createStarted(); + 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()) { + return; + } + } + + Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); + } + + fail("Member(s) " + otherMembersSet + " are not Up"); + } + + public static ActorRef findLocalShard(ActorContext actorContext, String shardName) { ActorRef shard = null; for(int i = 0; i < 20 * 5 && shard == null; i++) { Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); @@ -103,6 +147,31 @@ class IntegrationTestKit extends ShardTestKit { return shard; } + public static void verifyShardStats(DistributedDataStore datastore, String shardName, ShardStatsVerifier 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) { + ShardStats shardStats = (ShardStats)actorContext. + executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE); + + try { + verifier.verify(shardStats); + return; + } catch (AssertionError e) { + lastError = e; + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + } + } + + throw lastError; + } + void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath, NormalizedNode nodeToWrite) throws Exception { @@ -132,7 +201,7 @@ class IntegrationTestKit extends ShardTestKit { assertEquals("Data node", nodeToWrite, optional.get()); } - void doCommit(final DOMStoreThreePhaseCommitCohort cohort) throws Exception { + public void doCommit(final DOMStoreThreePhaseCommitCohort cohort) throws Exception { Boolean canCommit = cohort.canCommit().get(7, TimeUnit.SECONDS); assertEquals("canCommit", true, canCommit); cohort.preCommit().get(5, TimeUnit.SECONDS); @@ -146,12 +215,6 @@ class IntegrationTestKit extends ShardTestKit { cohort.commit().get(5, TimeUnit.SECONDS); } - void cleanup(DistributedDataStore dataStore) { - if(dataStore != null) { - dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), null); - } - } - void assertExceptionOnCall(Callable callable, Class expType) throws Exception { try { @@ -188,4 +251,8 @@ class IntegrationTestKit extends ShardTestKit { } }, expType); } -} \ No newline at end of file + + public interface ShardStatsVerifier { + void verify(ShardStats stats); + } +}