Bug 3206: CDS - issues with direct commit
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / DistributedDataStoreRemotingIntegrationTest.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 akka.actor.ActorRef;
13 import akka.actor.ActorSystem;
14 import akka.actor.Address;
15 import akka.actor.AddressFromURIString;
16 import akka.cluster.Cluster;
17 import akka.testkit.JavaTestKit;
18 import com.google.common.base.Optional;
19 import com.typesafe.config.ConfigFactory;
20 import java.math.BigInteger;
21 import java.util.concurrent.TimeUnit;
22 import org.junit.After;
23 import org.junit.Before;
24 import org.junit.Test;
25 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
26 import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
27 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
28 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
29 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
30 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
31 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
32 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
33 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
34 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
35 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
36 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
37 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
38 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
39 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
40 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
41 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
42 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
43 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
44 import org.opendaylight.yangtools.yang.data.api.schema.tree.TipProducingDataTree;
45 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
46 import org.opendaylight.yangtools.yang.data.impl.schema.builder.api.CollectionNodeBuilder;
47 import org.opendaylight.yangtools.yang.data.impl.schema.tree.InMemoryDataTreeFactory;
48
49 /**
50  * End-to-end distributed data store tests that exercise remote shards and transactions.
51  *
52  * @author Thomas Pantelis
53  */
54 public class DistributedDataStoreRemotingIntegrationTest {
55
56     private static final String[] SHARD_NAMES = {"cars", "people"};
57
58     private ActorSystem leaderSystem;
59     private ActorSystem followerSystem;
60
61     private final DatastoreContext.Builder leaderDatastoreContextBuilder =
62             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(1);
63
64     private final DatastoreContext.Builder followerDatastoreContextBuilder =
65             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(200);
66
67     private DistributedDataStore followerDistributedDataStore;
68     private DistributedDataStore leaderDistributedDataStore;
69     private IntegrationTestKit followerTestKit;
70     private IntegrationTestKit leaderTestKit;
71
72     @Before
73     public void setUpClass() {
74         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
75         Address member1Address = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
76         Cluster.get(leaderSystem).join(member1Address);
77
78         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
79         Cluster.get(followerSystem).join(member1Address);
80     }
81
82     @After
83     public void tearDownClass() {
84         JavaTestKit.shutdownActorSystem(leaderSystem);
85         JavaTestKit.shutdownActorSystem(followerSystem);
86     }
87
88     private void initDatastores(String type) {
89         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
90
91         String moduleShardsConfig = "module-shards-member1-and-2.conf";
92
93         followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
94         followerDistributedDataStore = followerTestKit.setupDistributedDataStore(type, moduleShardsConfig, false, SHARD_NAMES);
95
96         leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(type, moduleShardsConfig, true, SHARD_NAMES);
97
98         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), SHARD_NAMES);
99     }
100
101     private void verifyCars(DOMStoreReadTransaction readTx, MapEntryNode... entries) throws Exception {
102         Optional<NormalizedNode<?, ?>> optional = readTx.read(CarsModel.CAR_LIST_PATH).get(5, TimeUnit.SECONDS);
103         assertEquals("isPresent", true, optional.isPresent());
104
105         CollectionNodeBuilder<MapEntryNode, MapNode> listBuilder = ImmutableNodes.mapNodeBuilder(CarsModel.CAR_QNAME);
106         for(NormalizedNode<?, ?> entry: entries) {
107             listBuilder.withChild((MapEntryNode) entry);
108         }
109
110         assertEquals("Car list node", listBuilder.build(), optional.get());
111     }
112
113     private void verifyNode(DOMStoreReadTransaction readTx, YangInstanceIdentifier path, NormalizedNode<?, ?> expNode)
114             throws Exception {
115         Optional<NormalizedNode<?, ?>> optional = readTx.read(path).get(5, TimeUnit.SECONDS);
116         assertEquals("isPresent", true, optional.isPresent());
117         assertEquals("Data node", expNode, optional.get());
118     }
119
120     private void verifyExists(DOMStoreReadTransaction readTx, YangInstanceIdentifier path) throws Exception {
121         Boolean exists = readTx.exists(path).get(5, TimeUnit.SECONDS);
122         assertEquals("exists", true, exists);
123     }
124
125     @Test
126     public void testWriteTransactionWithSingleShard() throws Exception {
127         String testName = "testWriteTransactionWithSingleShard";
128         initDatastores(testName);
129
130         String followerCarShardName = "member-2-shard-cars-" + testName;
131         InMemoryJournal.addWriteMessagesCompleteLatch(followerCarShardName, 2, ApplyJournalEntries.class );
132
133         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
134         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
135
136         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
137         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
138
139         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
140         YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
141         writeTx.merge(car1Path, car1);
142
143         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
144         YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
145         writeTx.merge(car2Path, car2);
146
147         followerTestKit.doCommit(writeTx.ready());
148
149         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
150
151         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car1, car2);
152
153         // Test delete
154
155         writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
156
157         writeTx.delete(car1Path);
158
159         followerTestKit.doCommit(writeTx.ready());
160
161         verifyExists(followerDistributedDataStore.newReadOnlyTransaction(), car2Path);
162
163         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
164
165         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car2);
166
167         // Re-instate the follower member 2 as a single-node to verify replication and recovery.
168
169         InMemoryJournal.waitForWriteMessagesComplete(followerCarShardName);
170
171         JavaTestKit.shutdownActorSystem(leaderSystem, null, true);
172         JavaTestKit.shutdownActorSystem(followerSystem, null, true);
173
174         ActorSystem newSystem = ActorSystem.create("reinstated-member2", ConfigFactory.load().getConfig("Member2"));
175
176         DistributedDataStore member2Datastore = new IntegrationTestKit(newSystem, leaderDatastoreContextBuilder).
177                 setupDistributedDataStore(testName, "module-shards-member2", true, SHARD_NAMES);
178
179         verifyCars(member2Datastore.newReadOnlyTransaction(), car2);
180
181         JavaTestKit.shutdownActorSystem(newSystem);
182     }
183
184     @Test
185     public void testReadWriteTransactionWithSingleShard() throws Exception {
186         initDatastores("testReadWriteTransactionWithSingleShard");
187
188         DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
189         assertNotNull("newReadWriteTransaction returned null", rwTx);
190
191         rwTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
192         rwTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
193
194         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
195         rwTx.merge(CarsModel.newCarPath("optima"), car1);
196
197         verifyCars(rwTx, car1);
198
199         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
200         YangInstanceIdentifier car2Path = CarsModel.newCarPath("sportage");
201         rwTx.merge(car2Path, car2);
202
203         verifyExists(rwTx, car2Path);
204
205         followerTestKit.doCommit(rwTx.ready());
206
207         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car1, car2);
208     }
209
210     @Test
211     public void testWriteTransactionWithMultipleShards() throws Exception {
212         initDatastores("testWriteTransactionWithMultipleShards");
213
214         DOMStoreWriteTransaction writeTx = followerDistributedDataStore.newWriteOnlyTransaction();
215         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
216
217         YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
218         NormalizedNode<?, ?> carsNode = CarsModel.emptyContainer();
219         writeTx.write(carsPath, carsNode);
220
221         YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
222         NormalizedNode<?, ?> peopleNode = PeopleModel.emptyContainer();
223         writeTx.write(peoplePath, peopleNode);
224
225         followerTestKit.doCommit(writeTx.ready());
226
227         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
228
229         verifyNode(readTx, carsPath, carsNode);
230         verifyNode(readTx, peoplePath, peopleNode);
231     }
232
233     @Test
234     public void testReadWriteTransactionWithMultipleShards() throws Exception {
235         initDatastores("testReadWriteTransactionWithMultipleShards");
236
237         DOMStoreReadWriteTransaction rwTx = followerDistributedDataStore.newReadWriteTransaction();
238         assertNotNull("newReadWriteTransaction returned null", rwTx);
239
240         YangInstanceIdentifier carsPath = CarsModel.BASE_PATH;
241         NormalizedNode<?, ?> carsNode = CarsModel.emptyContainer();
242         rwTx.write(carsPath, carsNode);
243
244         YangInstanceIdentifier peoplePath = PeopleModel.BASE_PATH;
245         NormalizedNode<?, ?> peopleNode = PeopleModel.emptyContainer();
246         rwTx.write(peoplePath, peopleNode);
247
248         followerTestKit.doCommit(rwTx.ready());
249
250         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
251
252         verifyNode(readTx, carsPath, carsNode);
253         verifyNode(readTx, peoplePath, peopleNode);
254     }
255
256     @Test
257     public void testTransactionChainWithSingleShard() throws Exception {
258         initDatastores("testTransactionChainWithSingleShard");
259
260         DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
261
262         // Add the top-level cars container with write-only.
263
264         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
265         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
266
267         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
268
269         writeTx.ready();
270
271         // Verify the top-level cars container with read-only.
272
273         verifyNode(txChain.newReadOnlyTransaction(), CarsModel.BASE_PATH, CarsModel.emptyContainer());
274
275         // Perform car operations with read-write.
276
277         DOMStoreReadWriteTransaction rwTx = txChain.newReadWriteTransaction();
278
279         verifyNode(rwTx, CarsModel.BASE_PATH, CarsModel.emptyContainer());
280
281         rwTx.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
282
283         MapEntryNode car1 = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
284         YangInstanceIdentifier car1Path = CarsModel.newCarPath("optima");
285         rwTx.write(car1Path, car1);
286
287         verifyExists(rwTx, car1Path);
288
289         verifyCars(rwTx, car1);
290
291         MapEntryNode car2 = CarsModel.newCarEntry("sportage", BigInteger.valueOf(25000));
292         rwTx.merge(CarsModel.newCarPath("sportage"), car2);
293
294         rwTx.delete(car1Path);
295
296         followerTestKit.doCommit(rwTx.ready());
297
298         txChain.close();
299
300         verifyCars(followerDistributedDataStore.newReadOnlyTransaction(), car2);
301     }
302
303     @Test
304     public void testTransactionChainWithMultipleShards() throws Exception{
305         initDatastores("testTransactionChainWithMultipleShards");
306
307         DOMStoreTransactionChain txChain = followerDistributedDataStore.createTransactionChain();
308
309         DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
310         assertNotNull("newWriteOnlyTransaction returned null", writeTx);
311
312         writeTx.write(CarsModel.BASE_PATH, CarsModel.emptyContainer());
313         writeTx.write(PeopleModel.BASE_PATH, PeopleModel.emptyContainer());
314
315         writeTx.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
316         writeTx.write(PeopleModel.PERSON_LIST_PATH, PeopleModel.newPersonMapNode());
317
318         followerTestKit.doCommit(writeTx.ready());
319
320         DOMStoreReadWriteTransaction readWriteTx = txChain.newReadWriteTransaction();
321
322         MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
323         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
324         readWriteTx.write(carPath, car);
325
326         MapEntryNode person = PeopleModel.newPersonEntry("jack");
327         YangInstanceIdentifier personPath = PeopleModel.newPersonPath("jack");
328         readWriteTx.merge(personPath, person);
329
330         Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(carPath).get(5, TimeUnit.SECONDS);
331         assertEquals("isPresent", true, optional.isPresent());
332         assertEquals("Data node", car, optional.get());
333
334         optional = readWriteTx.read(personPath).get(5, TimeUnit.SECONDS);
335         assertEquals("isPresent", true, optional.isPresent());
336         assertEquals("Data node", person, optional.get());
337
338         DOMStoreThreePhaseCommitCohort cohort2 = readWriteTx.ready();
339
340         writeTx = txChain.newWriteOnlyTransaction();
341
342         writeTx.delete(personPath);
343
344         DOMStoreThreePhaseCommitCohort cohort3 = writeTx.ready();
345
346         followerTestKit.doCommit(cohort2);
347         followerTestKit.doCommit(cohort3);
348
349         txChain.close();
350
351         DOMStoreReadTransaction readTx = followerDistributedDataStore.newReadOnlyTransaction();
352         verifyCars(readTx, car);
353
354         optional = readTx.read(personPath).get(5, TimeUnit.SECONDS);
355         assertEquals("isPresent", false, optional.isPresent());
356     }
357
358     @Test
359     public void testReadyLocalTransactionForwardedToLeader() throws Exception {
360         initDatastores("testReadyLocalTransactionForwardedToLeader");
361
362         Optional<ActorRef> carsFollowerShard = followerDistributedDataStore.getActorContext().findLocalShard("cars");
363         assertEquals("Cars follower shard found", true, carsFollowerShard.isPresent());
364
365         TipProducingDataTree dataTree = InMemoryDataTreeFactory.getInstance().create();
366         dataTree.setSchemaContext(SchemaContextHelper.full());
367         DataTreeModification modification = dataTree.takeSnapshot().newModification();
368
369         new WriteModification(CarsModel.BASE_PATH, CarsModel.emptyContainer()).apply(modification);
370         new MergeModification(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()).apply(modification);
371
372         MapEntryNode car = CarsModel.newCarEntry("optima", BigInteger.valueOf(20000));
373         new WriteModification(CarsModel.newCarPath("optima"), car).apply(modification);
374
375         String transactionID = "tx-1";
376         ReadyLocalTransaction readyLocal = new ReadyLocalTransaction(transactionID , modification, true);
377
378         carsFollowerShard.get().tell(readyLocal, followerTestKit.getRef());
379         followerTestKit.expectMsgClass(CommitTransactionReply.SERIALIZABLE_CLASS);
380
381         verifyCars(leaderDistributedDataStore.newReadOnlyTransaction(), car);
382     }
383 }

©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.