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 }