opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/DistributedDataStoreIntegrationTest.java

   1 package org.opendaylight.controller.cluster.datastore;
   2
   3 import akka.actor.ActorRef;
   4 import akka.actor.ActorSystem;
   5 import akka.actor.PoisonPill;
   6 import com.google.common.base.Optional;
   7 import com.google.common.util.concurrent.Uninterruptibles;
   8 import static org.junit.Assert.assertEquals;
   9 import static org.junit.Assert.assertNotNull;
  10 import org.junit.Test;
  11 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  12 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
  13 import org.opendaylight.controller.cluster.datastore.utils.MockDataChangeListener;
  14 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
  15 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
  16 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  17 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  18 import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
  19 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  20 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadWriteTransaction;
  21 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
  22 import org.opendaylight.controller.sal.core.spi.data.DOMStoreTransactionChain;
  23 import org.opendaylight.controller.sal.core.spi.data.DOMStoreWriteTransaction;
  24 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  25 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  26 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  27 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  28 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  29 import java.util.concurrent.TimeUnit;
  30
  31 public class DistributedDataStoreIntegrationTest extends AbstractActorTest {
  32
  33     @Test
  34     public void testWriteTransactionWithSingleShard() throws Exception{
  35         System.setProperty("shard.persistent", "true");
  36         new IntegrationTestKit(getSystem()) {{
  37             DistributedDataStore dataStore =
  38                     setupDistributedDataStore("transactionIntegrationTest", "test-1");
  39
  40             testWriteTransaction(dataStore, TestModel.TEST_PATH,
  41                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
  42
  43             testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
  44                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
  45
  46             cleanup(dataStore);
  47         }};
  48     }
  49
  50     @Test
  51     public void testWriteTransactionWithMultipleShards() throws Exception{
  52         System.setProperty("shard.persistent", "true");
  53         new IntegrationTestKit(getSystem()) {{
  54             DistributedDataStore dataStore =
  55                     setupDistributedDataStore("testWriteTransactionWithMultipleShards", "cars-1", "people-1");
  56
  57             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
  58             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
  59
  60             YangInstanceIdentifier nodePath1 = CarsModel.BASE_PATH;
  61             NormalizedNode<?, ?> nodeToWrite1 = CarsModel.emptyContainer();
  62             writeTx.write(nodePath1, nodeToWrite1);
  63
  64             YangInstanceIdentifier nodePath2 = PeopleModel.BASE_PATH;
  65             NormalizedNode<?, ?> nodeToWrite2 = PeopleModel.emptyContainer();
  66             writeTx.write(nodePath2, nodeToWrite2);
  67
  68             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
  69
  70             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
  71             assertEquals("canCommit", true, canCommit);
  72             cohort.preCommit().get(5, TimeUnit.SECONDS);
  73             cohort.commit().get(5, TimeUnit.SECONDS);
  74
  75             // 5. Verify the data in the store
  76
  77             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
  78
  79             Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath1).get();
  80             assertEquals("isPresent", true, optional.isPresent());
  81             assertEquals("Data node", nodeToWrite1, optional.get());
  82
  83             optional = readTx.read(nodePath2).get();
  84             assertEquals("isPresent", true, optional.isPresent());
  85             assertEquals("Data node", nodeToWrite2, optional.get());
  86
  87             cleanup(dataStore);
  88         }};
  89     }
  90
  91     @Test
  92     public void testReadWriteTransaction() throws Exception{
  93         System.setProperty("shard.persistent", "true");
  94         new IntegrationTestKit(getSystem()) {{
  95             DistributedDataStore dataStore =
  96                     setupDistributedDataStore("testReadWriteTransaction", "test-1");
  97
  98          // 1. Create a read-write Tx
  99
 100             DOMStoreReadWriteTransaction readWriteTx = dataStore.newReadWriteTransaction();
 101             assertNotNull("newReadWriteTransaction returned null", readWriteTx);
 102
 103             // 2. Write some data
 104
 105             YangInstanceIdentifier nodePath = TestModel.TEST_PATH;
 106             NormalizedNode<?, ?> nodeToWrite = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 107             readWriteTx.write(nodePath, nodeToWrite );
 108
 109             // 3. Read the data from Tx
 110
 111             Boolean exists = readWriteTx.exists(nodePath).checkedGet(5, TimeUnit.SECONDS);
 112             assertEquals("exists", true, exists);
 113
 114             Optional<NormalizedNode<?, ?>> optional = readWriteTx.read(nodePath).get(5, TimeUnit.SECONDS);
 115             assertEquals("isPresent", true, optional.isPresent());
 116             assertEquals("Data node", nodeToWrite, optional.get());
 117
 118             // 4. Ready the Tx for commit
 119
 120             DOMStoreThreePhaseCommitCohort cohort = readWriteTx.ready();
 121
 122             // 5. Commit the Tx
 123
 124             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 125             assertEquals("canCommit", true, canCommit);
 126             cohort.preCommit().get(5, TimeUnit.SECONDS);
 127             cohort.commit().get(5, TimeUnit.SECONDS);
 128
 129             // 6. Verify the data in the store
 130
 131             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 132
 133             optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
 134             assertEquals("isPresent", true, optional.isPresent());
 135             assertEquals("Data node", nodeToWrite, optional.get());
 136
 137             cleanup(dataStore);
 138         }};
 139     }
 140
 141     @Test
 142     public void testTransactionAbort() throws Exception{
 143         System.setProperty("shard.persistent", "true");
 144         new IntegrationTestKit(getSystem()) {{
 145             DistributedDataStore dataStore =
 146                     setupDistributedDataStore("transactionAbortIntegrationTest", "test-1");
 147
 148             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 149             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 150
 151             writeTx.write(TestModel.TEST_PATH, ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 152
 153             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 154
 155             cohort.canCommit().get(5, TimeUnit.SECONDS);
 156
 157             cohort.abort().get(5, TimeUnit.SECONDS);
 158
 159             testWriteTransaction(dataStore, TestModel.TEST_PATH,
 160                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 161
 162             cleanup(dataStore);
 163         }};
 164     }
 165
 166     @Test
 167     public void testTransactionChain() throws Exception{
 168         System.setProperty("shard.persistent", "true");
 169         new IntegrationTestKit(getSystem()) {{
 170             DistributedDataStore dataStore =
 171                     setupDistributedDataStore("transactionChainIntegrationTest", "test-1");
 172
 173             // 1. Create a Tx chain and write-only Tx
 174
 175             DOMStoreTransactionChain txChain = dataStore.createTransactionChain();
 176
 177             DOMStoreWriteTransaction writeTx = txChain.newWriteOnlyTransaction();
 178             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 179
 180             // 2. Write some data
 181
 182             NormalizedNode<?, ?> containerNode = ImmutableNodes.containerNode(TestModel.TEST_QNAME);
 183             writeTx.write(TestModel.TEST_PATH, containerNode);
 184
 185             // 3. Ready the Tx for commit
 186
 187             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 188
 189             // 4. Commit the Tx
 190
 191             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 192             assertEquals("canCommit", true, canCommit);
 193             cohort.preCommit().get(5, TimeUnit.SECONDS);
 194             cohort.commit().get(5, TimeUnit.SECONDS);
 195
 196             // 5. Verify the data in the store
 197
 198             DOMStoreReadTransaction readTx = txChain.newReadOnlyTransaction();
 199
 200             Optional<NormalizedNode<?, ?>> optional = readTx.read(TestModel.TEST_PATH).get(5, TimeUnit.SECONDS);
 201             assertEquals("isPresent", true, optional.isPresent());
 202             assertEquals("Data node", containerNode, optional.get());
 203
 204             txChain.close();
 205
 206             cleanup(dataStore);
 207         }};
 208     }
 209
 210     @Test
 211     public void testChangeListenerRegistration() throws Exception{
 212         new IntegrationTestKit(getSystem()) {{
 213             DistributedDataStore dataStore =
 214                     setupDistributedDataStore("testChangeListenerRegistration", "test-1");
 215
 216             MockDataChangeListener listener = new MockDataChangeListener(3);
 217
 218             ListenerRegistration<MockDataChangeListener>
 219                     listenerReg = dataStore.registerChangeListener(TestModel.TEST_PATH, listener,
 220                             DataChangeScope.SUBTREE);
 221
 222             assertNotNull("registerChangeListener returned null", listenerReg);
 223
 224             testWriteTransaction(dataStore, TestModel.TEST_PATH,
 225                     ImmutableNodes.containerNode(TestModel.TEST_QNAME));
 226
 227             testWriteTransaction(dataStore, TestModel.OUTER_LIST_PATH,
 228                     ImmutableNodes.mapNodeBuilder(TestModel.OUTER_LIST_QNAME).build());
 229
 230             YangInstanceIdentifier listPath = YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
 231                     nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1).build();
 232             testWriteTransaction(dataStore, listPath,
 233                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 1));
 234
 235             listener.waitForChangeEvents(TestModel.TEST_PATH, TestModel.OUTER_LIST_PATH, listPath );
 236
 237             listenerReg.close();
 238
 239             testWriteTransaction(dataStore, YangInstanceIdentifier.builder(TestModel.OUTER_LIST_PATH).
 240                     nodeWithKey(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2).build(),
 241                     ImmutableNodes.mapEntry(TestModel.OUTER_LIST_QNAME, TestModel.ID_QNAME, 2));
 242
 243             listener.expectNoMoreChanges("Received unexpected change after close");
 244
 245             cleanup(dataStore);
 246         }};
 247     }
 248
 249     class IntegrationTestKit extends ShardTestKit {
 250
 251         IntegrationTestKit(ActorSystem actorSystem) {
 252             super(actorSystem);
 253         }
 254
 255         DistributedDataStore setupDistributedDataStore(String typeName, String... shardNames) {
 256             MockClusterWrapper cluster = new MockClusterWrapper();
 257             Configuration config = new ConfigurationImpl("module-shards.conf", "modules.conf");
 258             ShardStrategyFactory.setConfiguration(config);
 259
 260             DatastoreContext datastoreContext = DatastoreContext.newBuilder().build();
 261             DistributedDataStore dataStore = new DistributedDataStore(getSystem(), typeName, cluster,
 262                     config, datastoreContext);
 263
 264             SchemaContext schemaContext = SchemaContextHelper.full();
 265             dataStore.onGlobalContextUpdated(schemaContext);
 266
 267             for(String shardName: shardNames) {
 268                 ActorRef shard = null;
 269                 for(int i = 0; i < 20 * 5 && shard == null; i++) {
 270                     Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 271                     Optional<ActorRef> shardReply = dataStore.getActorContext().findLocalShard(shardName);
 272                     if(shardReply.isPresent()) {
 273                         shard = shardReply.get();
 274                     }
 275                 }
 276
 277                 assertNotNull("Shard was not created", shard);
 278
 279                 System.out.println("!!!!!!shard: "+shard.path().toString());
 280                 waitUntilLeader(shard);
 281             }
 282
 283             return dataStore;
 284         }
 285
 286         void testWriteTransaction(DistributedDataStore dataStore, YangInstanceIdentifier nodePath,
 287                 NormalizedNode<?, ?> nodeToWrite) throws Exception {
 288
 289             // 1. Create a write-only Tx
 290
 291             DOMStoreWriteTransaction writeTx = dataStore.newWriteOnlyTransaction();
 292             assertNotNull("newWriteOnlyTransaction returned null", writeTx);
 293
 294             // 2. Write some data
 295
 296             writeTx.write(nodePath, nodeToWrite);
 297
 298             // 3. Ready the Tx for commit
 299
 300             DOMStoreThreePhaseCommitCohort cohort = writeTx.ready();
 301
 302             // 4. Commit the Tx
 303
 304             Boolean canCommit = cohort.canCommit().get(5, TimeUnit.SECONDS);
 305             assertEquals("canCommit", true, canCommit);
 306             cohort.preCommit().get(5, TimeUnit.SECONDS);
 307             cohort.commit().get(5, TimeUnit.SECONDS);
 308
 309             // 5. Verify the data in the store
 310
 311             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 312
 313             Optional<NormalizedNode<?, ?>> optional = readTx.read(nodePath).get(5, TimeUnit.SECONDS);
 314             assertEquals("isPresent", true, optional.isPresent());
 315             assertEquals("Data node", nodeToWrite, optional.get());
 316         }
 317
 318         void cleanup(DistributedDataStore dataStore) {
 319             dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), null);
 320         }
 321     }
 322
 323 }