BUG 2138: Introduce prefix based shards into ShardManager
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / IntegrationTestKit.java
1 /*
2  * Copyright (c) 2015 Brocade Communications Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.controller.cluster.datastore;
9
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertNotNull;
12 import static org.junit.Assert.fail;
13
14 import akka.actor.ActorRef;
15 import akka.actor.ActorSystem;
16 import akka.cluster.Cluster;
17 import akka.cluster.ClusterEvent.CurrentClusterState;
18 import akka.cluster.Member;
19 import akka.cluster.MemberStatus;
20 import com.google.common.base.Optional;
21 import com.google.common.base.Stopwatch;
22 import com.google.common.collect.Sets;
23 import com.google.common.util.concurrent.ListenableFuture;
24 import com.google.common.util.concurrent.Uninterruptibles;
25 import java.util.Set;
26 import java.util.concurrent.Callable;
27 import java.util.concurrent.TimeUnit;
28 import org.mockito.Mockito;
29 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
30 import org.opendaylight.controller.cluster.datastore.config.Configuration;
31 import org.opendaylight.controller.cluster.datastore.config.ConfigurationImpl;
32 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
33 import org.opendaylight.controller.cluster.datastore.messages.DatastoreSnapshot;
34 import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
35 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
36 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
37 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
38 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
39 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
40 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
41 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
42 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
43 import scala.concurrent.Await;
44 import scala.concurrent.Future;
45 import scala.concurrent.duration.Duration;
46
47 public class IntegrationTestKit extends ShardTestKit {
48
49     protected DatastoreContext.Builder datastoreContextBuilder;
50     protected DatastoreSnapshot restoreFromSnapshot;
51
52     public IntegrationTestKit(ActorSystem actorSystem, Builder datastoreContextBuilder) {
53         super(actorSystem);
54         this.datastoreContextBuilder = datastoreContextBuilder;
55     }
56
57     public DatastoreContext.Builder getDatastoreContextBuilder() {
58         return datastoreContextBuilder;
59     }
60
61     public DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) {
62         return setupDistributedDataStore(typeName, "module-shards.conf", true, SchemaContextHelper.full(), shardNames);
63     }
64
65     public DistributedDataStore setupDistributedDataStore(String typeName, boolean waitUntilLeader,
66             String... shardNames) {
67         return setupDistributedDataStore(typeName, "module-shards.conf", waitUntilLeader,
68                 SchemaContextHelper.full(), shardNames);
69     }
70
71     public DistributedDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig,
72             final boolean waitUntilLeader, final String... shardNames) {
73         return setupDistributedDataStore(typeName, moduleShardsConfig, waitUntilLeader,
74                 SchemaContextHelper.full(), shardNames);
75     }
76
77     public DistributedDataStore setupDistributedDataStore(final String typeName, final String moduleShardsConfig,
78             final boolean waitUntilLeader, final SchemaContext schemaContext, final String... shardNames) {
79         final ClusterWrapper cluster = new ClusterWrapperImpl(getSystem());
80         final Configuration config = new ConfigurationImpl(moduleShardsConfig, "modules.conf");
81
82         datastoreContextBuilder.dataStoreName(typeName);
83
84         DatastoreContext datastoreContext = datastoreContextBuilder.build();
85         DatastoreContextFactory mockContextFactory = Mockito.mock(DatastoreContextFactory.class);
86         Mockito.doReturn(datastoreContext).when(mockContextFactory).getBaseDatastoreContext();
87         Mockito.doReturn(datastoreContext).when(mockContextFactory).getShardDatastoreContext(Mockito.anyString());
88
89         DistributedDataStore dataStore = new DistributedDataStore(getSystem(), cluster, config, mockContextFactory,
90                 restoreFromSnapshot);
91
92         dataStore.onGlobalContextUpdated(schemaContext);
93
94         if (waitUntilLeader) {
95             waitUntilLeader(dataStore.getActorContext(), shardNames);
96         }
97
98         datastoreContextBuilder = DatastoreContext.newBuilderFrom(datastoreContext);
99         return dataStore;
100     }
101
102     public void waitUntilLeader(ActorContext actorContext, String... shardNames) {
103         for (String shardName: shardNames) {
104             ActorRef shard = findLocalShard(actorContext, shardName);
105
106             assertNotNull("Shard was not created for " + shardName, shard);
107
108             waitUntilLeader(shard);
109         }
110     }
111
112     public void waitUntilNoLeader(ActorContext actorContext, String... shardNames) {
113         for (String shardName: shardNames) {
114             ActorRef shard = findLocalShard(actorContext, shardName);
115             assertNotNull("No local shard found for " + shardName, shard);
116
117             waitUntilNoLeader(shard);
118         }
119     }
120
121     public void waitForMembersUp(String... otherMembers) {
122         Set<String> otherMembersSet = Sets.newHashSet(otherMembers);
123         Stopwatch sw = Stopwatch.createStarted();
124         while (sw.elapsed(TimeUnit.SECONDS) <= 10) {
125             CurrentClusterState state = Cluster.get(getSystem()).state();
126             for (Member m: state.getMembers()) {
127                 if (m.status() == MemberStatus.up() && otherMembersSet.remove(m.getRoles().iterator().next())
128                         && otherMembersSet.isEmpty()) {
129                     return;
130                 }
131             }
132
133             Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
134         }
135
136         fail("Member(s) " + otherMembersSet + " are not Up");
137     }
138
139     public static ActorRef findLocalShard(ActorContext actorContext, String shardName) {
140         ActorRef shard = null;
141         for (int i = 0; i < 20 * 5 && shard == null; i++) {
142             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
143             Optional<ActorRef> shardReply = actorContext.findLocalShard(shardName);
144             if (shardReply.isPresent()) {
145                 shard = shardReply.get();
146             }
147         }
148         return shard;
149     }
150
151     public static void verifyShardStats(DistributedDataStore datastore, String shardName, ShardStatsVerifier verifier)
152             throws Exception {
153         ActorContext actorContext = datastore.getActorContext();
154
155         Future<ActorRef> future = actorContext.findLocalShardAsync(shardName);
156         ActorRef shardActor = Await.result(future, Duration.create(10, TimeUnit.SECONDS));
157
158         AssertionError lastError = null;
159         Stopwatch sw = Stopwatch.createStarted();
160         while (sw.elapsed(TimeUnit.SECONDS) <= 5) {
161             ShardStats shardStats = (ShardStats)actorContext
162                     .executeOperation(shardActor, Shard.GET_SHARD_MBEAN_MESSAGE);
163
164             try {
165                 verifier.verify(shardStats);
166                 return;
167             } catch (AssertionError e) {
168                 lastError = e;
169                 Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
170             }
171         }
172
173         throw lastError;
174     }
175
176     void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath,
177             NormalizedNode<?, ?> nodeToWrite) throws Exception {
178
179         // 1. Create a write-only Tx
180
181         DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
182         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
183
184         // 2. Write some data
185
186         writeTx.write(nodePath, nodeToWrite);
187
188         // 3. Ready the Tx for commit
189
190         DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
191
192         // 4. Commit the Tx
193
194         doCommit(cohort);
195
196         // 5. Verify the data in the store
197
198         DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
199
200         Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
201         assertEquals("isPresent", true, optional.isPresent());
202         assertEquals("Data node", nodeToWrite, optional.get());
203     }
204
205     public void doCommit(final DOMStoreThreePhaseCommitCohort cohort) throws Exception {
206         Boolean canCommit = cohort.canCommit().get(7, TimeUnit.SECONDS);
207         assertEquals("canCommit", true, canCommit);
208         cohort.preCommit().get(5, TimeUnit.SECONDS);
209         cohort.commit().get(5, TimeUnit.SECONDS);
210     }
211
212     void doCommit(final ListenableFuture<Boolean> canCommitFuture, final DOMStoreThreePhaseCommitCohort cohort)
213             throws Exception {
214         Boolean canCommit = canCommitFuture.get(7, TimeUnit.SECONDS);
215         assertEquals("canCommit", true, canCommit);
216         cohort.preCommit().get(5, TimeUnit.SECONDS);
217         cohort.commit().get(5, TimeUnit.SECONDS);
218     }
219
220     @SuppressWarnings("checkstyle:IllegalCatch")
221     void assertExceptionOnCall(Callable<Void> callable, Class<? extends Exception> expType)
222             throws Exception {
223         try {
224             callable.call();
225             fail("Expected " + expType.getSimpleName());
226         } catch (Exception e) {
227             assertEquals("Exception type", expType, e.getClass());
228         }
229     }
230
231     void assertExceptionOnTxChainCreates(final DOMStoreTransactionChain txChain,
232             Class<? extends Exception> expType) throws Exception {
233         assertExceptionOnCall(new Callable<Void>() {
234             @Override
235             public Void call() throws Exception {
236                 txChain.newWriteOnlyTransaction();
237                 return null;
238             }
239         }, expType);
240
241         assertExceptionOnCall(() -> {
242             txChain.newReadWriteTransaction();
243             return null;
244         }, expType);
245
246         assertExceptionOnCall(() -> {
247             txChain.newReadOnlyTransaction();
248             return null;
249         }, expType);
250     }
251
252     public interface ShardStatsVerifier {
253         void verify(ShardStats stats);
254     }
255 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.