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

   1 /*
   2  * Copyright (c) 2016 Cisco 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
   9 package org.opendaylight.controller.cluster.sharding;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.mockito.Matchers.anyCollection;
  14 import static org.mockito.Matchers.anyMap;
  15 import static org.mockito.Mockito.doNothing;
  16 import static org.mockito.Mockito.doReturn;
  17 import static org.mockito.Mockito.mock;
  18 import static org.mockito.Mockito.timeout;
  19 import static org.mockito.Mockito.verify;
  20 import static org.mockito.Mockito.verifyNoMoreInteractions;
  21 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.findLocalShard;
  22 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.waitUntilShardIsDown;
  23
  24 import akka.actor.ActorRef;
  25 import akka.actor.ActorSystem;
  26 import akka.actor.Address;
  27 import akka.actor.AddressFromURIString;
  28 import akka.cluster.Cluster;
  29 import akka.testkit.JavaTestKit;
  30 import com.google.common.base.Optional;
  31 import com.google.common.collect.Lists;
  32 import com.google.common.util.concurrent.CheckedFuture;
  33 import com.typesafe.config.ConfigFactory;
  34 import java.util.ArrayList;
  35 import java.util.Collection;
  36 import java.util.Collections;
  37 import java.util.List;
  38 import java.util.Map;
  39 import org.junit.After;
  40 import org.junit.Assert;
  41 import org.junit.Before;
  42 import org.junit.Ignore;
  43 import org.junit.Test;
  44 import org.mockito.ArgumentCaptor;
  45 import org.mockito.Captor;
  46 import org.mockito.Mockito;
  47 import org.mockito.MockitoAnnotations;
  48 import org.opendaylight.controller.cluster.ActorSystemProvider;
  49 import org.opendaylight.controller.cluster.datastore.AbstractTest;
  50 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  51 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
  52 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  53 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
  54 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
  55 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
  56 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
  57 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
  58 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
  59 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
  60 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  61 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
  62 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  63 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  64 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
  65 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
  66 import org.opendaylight.mdsal.dom.api.DOMDataTreeListener;
  67 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
  68 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
  69 import org.opendaylight.yangtools.yang.common.QName;
  70 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  71 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
  72 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifierWithPredicates;
  73 import org.opendaylight.yangtools.yang.data.api.schema.LeafNode;
  74 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  75 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
  76 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  77 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  78 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  79 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
  80 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
  81 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableMapNodeBuilder;
  82 import org.slf4j.Logger;
  83 import org.slf4j.LoggerFactory;
  84
  85 @Ignore("distributed-data is broken needs to be removed")
  86 public class DistributedShardedDOMDataTreeTest extends AbstractTest {
  87
  88     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class);
  89
  90     private static final Address MEMBER_1_ADDRESS =
  91             AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
  92
  93     private static final DOMDataTreeIdentifier TEST_ID =
  94             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
  95
  96     private static final DOMDataTreeIdentifier INNER_LIST_ID =
  97             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
  98                     YangInstanceIdentifier.create(getOuterListIdFor(0).getPathArguments())
  99                             .node(TestModel.INNER_LIST_QNAME));
 100
 101     private ActorSystem leaderSystem;
 102
 103     private final Builder leaderDatastoreContextBuilder =
 104             DatastoreContext.newBuilder()
 105                     .shardHeartbeatIntervalInMillis(100)
 106                     .shardElectionTimeoutFactor(2)
 107                     .logicalStoreType(
 108                             org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType.CONFIGURATION);
 109
 110     private DistributedDataStore leaderDistributedDataStore;
 111     private IntegrationTestKit leaderTestKit;
 112
 113     private DistributedShardedDOMDataTree leaderShardFactory;
 114
 115     @Captor
 116     private ArgumentCaptor<Collection<DataTreeCandidate>> captorForChanges;
 117     @Captor
 118     private ArgumentCaptor<Map<DOMDataTreeIdentifier, NormalizedNode<?, ?>>> captorForSubtrees;
 119
 120     private ActorSystemProvider leaderSystemProvider;
 121
 122     @Before
 123     public void setUp() {
 124         MockitoAnnotations.initMocks(this);
 125
 126         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
 127         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
 128
 129         leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
 130         doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();
 131     }
 132
 133     @After
 134     public void tearDown() {
 135         if (leaderDistributedDataStore != null) {
 136             leaderDistributedDataStore.close();
 137         }
 138
 139         JavaTestKit.shutdownActorSystem(leaderSystem);
 140     }
 141
 142     private void initEmptyDatastore(final String type) {
 143         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
 144
 145         leaderDistributedDataStore =
 146                 leaderTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full());
 147
 148
 149         leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystemProvider,
 150                 leaderDistributedDataStore,
 151                 leaderDistributedDataStore);
 152     }
 153
 154
 155     @Test
 156     public void testWritesIntoDefaultShard() throws Exception {
 157         initEmptyDatastore("config");
 158
 159         leaderShardFactory.createDistributedShard(TEST_ID,
 160                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
 161
 162         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 163                 ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
 164
 165         final DOMDataTreeIdentifier configRoot =
 166                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
 167
 168         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(configRoot));
 169
 170         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
 171         final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
 172         Assert.assertNotNull(cursor);
 173     }
 174
 175     @Test
 176     public void testSingleNodeWrites() throws Exception {
 177         initEmptyDatastore("config");
 178
 179         leaderShardFactory.createDistributedShard(TEST_ID,
 180                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
 181
 182         final DistributedShardRegistration shardRegistration =
 183                 leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME));
 184         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 185                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 186
 187         LOG.warn("Got after waiting for nonleader");
 188         final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
 189
 190         new JavaTestKit(leaderSystem) {
 191             {
 192                 leaderShardManager.tell(
 193                         new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
 194                 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
 195
 196                 leaderDistributedDataStore.getActorContext().getShardManager().tell(
 197                         new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
 198                 expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
 199             }
 200         };
 201
 202         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
 203
 204         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
 205         final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
 206         Assert.assertNotNull(cursor);
 207         final YangInstanceIdentifier nameId =
 208                 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
 209         final LeafNode<String> valueToCheck = ImmutableLeafNodeBuilder.<String>create().withNodeIdentifier(
 210                 new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build();
 211         cursor.write(nameId.getLastPathArgument(),
 212                 valueToCheck);
 213
 214         cursor.close();
 215         LOG.warn("Got to pre submit");
 216
 217         tx.submit().checkedGet();
 218
 219         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 220         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 221
 222         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(TEST_ID),
 223                 true, Collections.emptyList());
 224
 225         verify(mockedDataTreeListener, timeout(1000).times(1)).onDataTreeChanged(captorForChanges.capture(),
 226                 captorForSubtrees.capture());
 227         final List<Collection<DataTreeCandidate>> capturedValue = captorForChanges.getAllValues();
 228
 229         final Optional<NormalizedNode<?, ?>> dataAfter =
 230                 capturedValue.get(0).iterator().next().getRootNode().getDataAfter();
 231
 232         final NormalizedNode<?,?> expected = ImmutableContainerNodeBuilder.create()
 233                 .withNodeIdentifier(new NodeIdentifier(TestModel.TEST_QNAME)).withChild(valueToCheck).build();
 234         assertEquals(expected, dataAfter.get());
 235
 236         verifyNoMoreInteractions(mockedDataTreeListener);
 237
 238     }
 239
 240     @Test
 241     public void testMultipleWritesIntoSingleMapEntry() throws Exception {
 242         initEmptyDatastore("config");
 243
 244         final DistributedShardRegistration shardRegistration =
 245                 leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME));
 246         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 247                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
 248
 249         LOG.warn("Got after waiting for nonleader");
 250         final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
 251
 252         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 253                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 254
 255         final YangInstanceIdentifier oid1 = getOuterListIdFor(0);
 256         final DOMDataTreeIdentifier outerListPath = new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1);
 257
 258         final DistributedShardRegistration outerListShardReg = leaderShardFactory.createDistributedShard(outerListPath,
 259                 Lists.newArrayList(AbstractTest.MEMBER_NAME));
 260
 261         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 262                 ClusterUtils.getCleanShardName(outerListPath.getRootIdentifier()));
 263
 264         final DOMDataTreeProducer shardProducer = leaderShardFactory.createProducer(
 265                 Collections.singletonList(outerListPath));
 266
 267         final DOMDataTreeCursorAwareTransaction tx = shardProducer.createTransaction(false);
 268         final DOMDataTreeWriteCursor cursor =
 269                 tx.createCursor(new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, oid1));
 270         assertNotNull(cursor);
 271
 272         MapNode innerList = ImmutableMapNodeBuilder
 273                 .create()
 274                 .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 275                 .build();
 276
 277         cursor.write(new NodeIdentifier(TestModel.INNER_LIST_QNAME), innerList);
 278         cursor.close();
 279         tx.submit().checkedGet();
 280
 281         final ArrayList<CheckedFuture<Void, TransactionCommitFailedException>> futures = new ArrayList<>();
 282         for (int i = 0; i < 1000; i++) {
 283             final Collection<MapEntryNode> innerListMapEntries = createInnerListMapEntries(1000, "run-" + i);
 284             for (final MapEntryNode innerListMapEntry : innerListMapEntries) {
 285                 final DOMDataTreeCursorAwareTransaction tx1 = shardProducer.createTransaction(false);
 286                 final DOMDataTreeWriteCursor cursor1 = tx1.createCursor(
 287                         new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION,
 288                                 oid1.node(new NodeIdentifier(TestModel.INNER_LIST_QNAME))));
 289                 cursor1.write(innerListMapEntry.getIdentifier(), innerListMapEntry);
 290                 cursor1.close();
 291                 futures.add(tx1.submit());
 292             }
 293         }
 294
 295         futures.get(futures.size() - 1).checkedGet();
 296
 297         final DOMDataTreeListener mockedDataTreeListener = mock(DOMDataTreeListener.class);
 298         doNothing().when(mockedDataTreeListener).onDataTreeChanged(anyCollection(), anyMap());
 299
 300         leaderShardFactory.registerListener(mockedDataTreeListener, Collections.singletonList(INNER_LIST_ID),
 301                 true, Collections.emptyList());
 302
 303         verify(mockedDataTreeListener, timeout(1000).times(1)).onDataTreeChanged(captorForChanges.capture(),
 304                 captorForSubtrees.capture());
 305         verifyNoMoreInteractions(mockedDataTreeListener);
 306         final List<Collection<DataTreeCandidate>> capturedValue = captorForChanges.getAllValues();
 307
 308         final NormalizedNode<?,?> expected =
 309                 ImmutableMapNodeBuilder
 310                         .create()
 311                         .withNodeIdentifier(new NodeIdentifier(TestModel.INNER_LIST_QNAME))
 312                                 // only the values from the last run should be present
 313                         .withValue(createInnerListMapEntries(1000, "run-999"))
 314                         .build();
 315
 316         assertEquals("List values dont match the expected values from the last run",
 317                 expected, capturedValue.get(0).iterator().next().getRootNode().getDataAfter().get());
 318
 319     }
 320
 321     private static Collection<MapEntryNode> createInnerListMapEntries(final int amount, final String valuePrefix) {
 322         final Collection<MapEntryNode> ret = new ArrayList<>();
 323         for (int i = 0; i < amount; i++) {
 324             ret.add(ImmutableNodes.mapEntryBuilder()
 325                     .withNodeIdentifier(new NodeIdentifierWithPredicates(TestModel.INNER_LIST_QNAME,
 326                             QName.create(TestModel.INNER_LIST_QNAME, "name"), Integer.toString(i)))
 327                     .withChild(ImmutableNodes
 328                             .leafNode(QName.create(TestModel.INNER_LIST_QNAME, "value"), valuePrefix + "-" + i))
 329                     .build());
 330         }
 331
 332         return ret;
 333     }
 334
 335     @Test
 336     public void testDistributedData() throws Exception {
 337         initEmptyDatastore("config");
 338
 339         leaderShardFactory.createDistributedShard(TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME));
 340         leaderShardFactory.createDistributedShard(
 341                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_CONTAINER_PATH),
 342                 Lists.newArrayList(AbstractTest.MEMBER_NAME));
 343         leaderShardFactory.createDistributedShard(
 344                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.INNER_LIST_PATH),
 345                 Lists.newArrayList(AbstractTest.MEMBER_NAME));
 346         leaderShardFactory.createDistributedShard(
 347                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.JUNK_PATH),
 348                 Lists.newArrayList(AbstractTest.MEMBER_NAME));
 349
 350         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 351                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 352         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 353                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
 354         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 355                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
 356         leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 357                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
 358
 359     }
 360
 361     @Test
 362     public void testMultipleRegistrationsAtOnePrefix() throws Exception {
 363         initEmptyDatastore("config");
 364
 365         for (int i = 0; i < 10; i++) {
 366             LOG.debug("Round {}", i);
 367             final DistributedShardRegistration reg1 = leaderShardFactory
 368                     .createDistributedShard(TEST_ID,
 369                             Lists.newArrayList(AbstractTest.MEMBER_NAME));
 370
 371             leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
 372                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 373
 374             assertNotNull(findLocalShard(leaderDistributedDataStore.getActorContext(),
 375                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
 376
 377             reg1.close();
 378
 379             waitUntilShardIsDown(leaderDistributedDataStore.getActorContext(),
 380                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
 381
 382         }
 383     }
 384
 385     private static YangInstanceIdentifier getOuterListIdFor(final int id) {
 386         return TestModel.OUTER_LIST_PATH.node(new NodeIdentifierWithPredicates(
 387                 TestModel.OUTER_LIST_QNAME, QName.create(TestModel.OUTER_LIST_QNAME, "id"), id));
 388     }
 389 }