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

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