Bug 8385 - @Ignore testMultipleRegistrationsAtOnePrefix
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / sharding / DistributedShardedDOMDataTreeRemotingTest.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.junit.Assert.assertTrue;
14 import static org.junit.Assert.fail;
15 import static org.mockito.Mockito.doReturn;
16 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.findLocalShard;
17 import static org.opendaylight.controller.cluster.datastore.IntegrationTestKit.waitUntilShardIsDown;
18
19 import akka.actor.ActorRef;
20 import akka.actor.ActorSystem;
21 import akka.actor.Address;
22 import akka.actor.AddressFromURIString;
23 import akka.cluster.Cluster;
24 import akka.testkit.JavaTestKit;
25 import com.google.common.collect.Lists;
26 import com.typesafe.config.ConfigFactory;
27 import java.util.Collections;
28 import java.util.HashSet;
29 import java.util.Set;
30 import org.junit.After;
31 import org.junit.Assert;
32 import org.junit.Before;
33 import org.junit.Ignore;
34 import org.junit.Test;
35 import org.mockito.Mockito;
36 import org.opendaylight.controller.cluster.ActorSystemProvider;
37 import org.opendaylight.controller.cluster.datastore.AbstractTest;
38 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
39 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
40 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
41 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
42 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
43 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
44 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
45 import org.opendaylight.controller.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
46 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
47 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
48 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
49 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
50 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
51 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
52 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
53 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
54 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
55 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
56 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
57 import org.slf4j.Logger;
58 import org.slf4j.LoggerFactory;
59
60 public class DistributedShardedDOMDataTreeRemotingTest extends AbstractTest {
61
62     private static final Logger LOG = LoggerFactory.getLogger(DistributedShardedDOMDataTreeRemotingTest.class);
63
64     private static final Address MEMBER_1_ADDRESS =
65             AddressFromURIString.parse("akka://cluster-test@127.0.0.1:2558");
66
67     private static final DOMDataTreeIdentifier TEST_ID =
68             new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
69
70     private static final String MODULE_SHARDS_CONFIG = "module-shards-default.conf";
71
72     private ActorSystem leaderSystem;
73     private ActorSystem followerSystem;
74
75
76     private final Builder leaderDatastoreContextBuilder =
77             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
78
79     private final DatastoreContext.Builder followerDatastoreContextBuilder =
80             DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
81
82     private DistributedDataStore leaderConfigDatastore;
83     private DistributedDataStore leaderOperDatastore;
84
85     private DistributedDataStore followerConfigDatastore;
86     private DistributedDataStore followerOperDatastore;
87
88
89     private IntegrationTestKit followerTestKit;
90     private IntegrationTestKit leaderTestKit;
91     private DistributedShardedDOMDataTree leaderShardFactory;
92
93     private DistributedShardedDOMDataTree followerShardFactory;
94     private ActorSystemProvider leaderSystemProvider;
95     private ActorSystemProvider followerSystemProvider;
96
97     @Before
98     public void setUp() {
99         InMemoryJournal.clear();
100         InMemorySnapshotStore.clear();
101
102         leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
103         Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
104
105         followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
106         Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
107
108         leaderSystemProvider = Mockito.mock(ActorSystemProvider.class);
109         doReturn(leaderSystem).when(leaderSystemProvider).getActorSystem();
110
111         followerSystemProvider = Mockito.mock(ActorSystemProvider.class);
112         doReturn(followerSystem).when(followerSystemProvider).getActorSystem();
113
114     }
115
116     @After
117     public void tearDown() {
118         if (leaderConfigDatastore != null) {
119             leaderConfigDatastore.close();
120         }
121         if (leaderOperDatastore != null) {
122             leaderOperDatastore.close();
123         }
124
125         if (followerConfigDatastore != null) {
126             followerConfigDatastore.close();
127         }
128         if (followerOperDatastore != null) {
129             followerOperDatastore.close();
130         }
131
132         JavaTestKit.shutdownActorSystem(leaderSystem, null, Boolean.TRUE);
133         JavaTestKit.shutdownActorSystem(followerSystem, null, Boolean.TRUE);
134
135         InMemoryJournal.clear();
136         InMemorySnapshotStore.clear();
137     }
138
139     private void initEmptyDatastores() throws Exception {
140         leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
141
142         leaderConfigDatastore = leaderTestKit.setupDistributedDataStore(
143                 "config", MODULE_SHARDS_CONFIG, true,
144                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
145         leaderOperDatastore = leaderTestKit.setupDistributedDataStore(
146                 "operational", MODULE_SHARDS_CONFIG, true,
147                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
148
149         leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystemProvider,
150                 leaderOperDatastore,
151                 leaderConfigDatastore);
152
153         followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
154
155         followerConfigDatastore = followerTestKit.setupDistributedDataStore(
156                 "config", MODULE_SHARDS_CONFIG, true, SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
157         followerOperDatastore = followerTestKit.setupDistributedDataStore(
158                 "operational", MODULE_SHARDS_CONFIG, true,
159                 SchemaContextHelper.distributedShardedDOMDataTreeSchemaContext());
160
161         followerShardFactory = new DistributedShardedDOMDataTree(followerSystemProvider,
162                 followerOperDatastore,
163                 followerConfigDatastore);
164
165         followerTestKit.waitForMembersUp("member-1");
166
167         leaderShardFactory.init();
168         followerShardFactory.init();
169
170         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
171                 ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
172
173         leaderTestKit.waitUntilLeader(leaderOperDatastore.getActorContext(),
174                 ClusterUtils.getCleanShardName(YangInstanceIdentifier.EMPTY));
175     }
176
177     @Test
178     public void testProducerRegistrations() throws Exception {
179         LOG.info("testProducerRegistrations starting");
180         initEmptyDatastores();
181
182         leaderTestKit.waitForMembersUp("member-2");
183
184         // TODO refactor shard creation and verification to own method
185         final DistributedShardRegistration shardRegistration =
186                 waitOnAsyncTask(leaderShardFactory.createDistributedShard(
187                         TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
188                         DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
189
190         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
191                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
192
193         final ActorRef leaderShardManager = leaderConfigDatastore.getActorContext().getShardManager();
194
195         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
196                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
197
198         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
199                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier())));
200
201         final Set<String> peers  = new HashSet<>();
202         IntegrationTestKit.verifyShardState(leaderConfigDatastore,
203                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
204                         peers.addAll(onDemandShardState.getPeerAddresses().values()));
205         assertEquals(peers.size(), 1);
206
207         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
208         try {
209             followerShardFactory.createProducer(Collections.singleton(TEST_ID));
210             fail("Producer should be already registered on the other node");
211         } catch (final IllegalArgumentException e) {
212             assertTrue(e.getMessage().contains("is attached to producer"));
213         }
214
215         producer.close();
216
217         final DOMDataTreeProducer followerProducer =
218                 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
219         try {
220             leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
221             fail("Producer should be already registered on the other node");
222         } catch (final IllegalArgumentException e) {
223             assertTrue(e.getMessage().contains("is attached to producer"));
224         }
225
226         followerProducer.close();
227         // try to create a shard on an already registered prefix on follower
228         try {
229             waitOnAsyncTask(followerShardFactory.createDistributedShard(
230                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
231                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
232             fail("This prefix already should have a shard registration that was forwarded from the other node");
233         } catch (final DOMDataTreeShardingConflictException e) {
234             assertTrue(e.getMessage().contains("is already occupied by another shard"));
235         }
236
237         shardRegistration.close().toCompletableFuture().get();
238
239         LOG.info("testProducerRegistrations ending");
240     }
241
242     @Test
243     public void testWriteIntoMultipleShards() throws Exception {
244         LOG.info("testWriteIntoMultipleShards starting");
245         initEmptyDatastores();
246
247         leaderTestKit.waitForMembersUp("member-2");
248
249         LOG.debug("registering first shard");
250         final DistributedShardRegistration shardRegistration =
251                 waitOnAsyncTask(leaderShardFactory.createDistributedShard(
252                         TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
253                         DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
254
255
256         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
257                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
258         findLocalShard(followerConfigDatastore.getActorContext(),
259                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
260
261         final Set<String> peers  = new HashSet<>();
262         IntegrationTestKit.verifyShardState(leaderConfigDatastore,
263                 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
264                         peers.addAll(onDemandShardState.getPeerAddresses().values()));
265         assertEquals(peers.size(), 1);
266
267         LOG.debug("Got after waiting for nonleader");
268         final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
269
270         final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
271         final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
272         Assert.assertNotNull(cursor);
273         final YangInstanceIdentifier nameId =
274                 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
275         cursor.write(nameId.getLastPathArgument(),
276                 ImmutableLeafNodeBuilder.<String>create().withNodeIdentifier(
277                         new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build());
278
279         cursor.close();
280         LOG.warn("Got to pre submit");
281
282         tx.submit().checkedGet();
283
284         shardRegistration.close().toCompletableFuture().get();
285
286         LOG.info("testWriteIntoMultipleShards ending");
287     }
288
289     @Test
290     public void testMultipleShardRegistrations() throws Exception {
291         LOG.info("testMultipleShardRegistrations starting");
292         initEmptyDatastores();
293
294         final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
295                 TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
296                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
297
298         final DistributedShardRegistration reg2 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
299                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.OUTER_CONTAINER_PATH),
300                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
301                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
302
303         final DistributedShardRegistration reg3 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
304                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.INNER_LIST_PATH),
305                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
306                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
307
308         final DistributedShardRegistration reg4 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
309                 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.JUNK_PATH),
310                 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
311                 DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
312
313         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
314                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
315         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
316                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
317         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
318                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
319         leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
320                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
321
322         // check leader has local shards
323         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
324                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
325
326         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
327                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
328
329         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
330                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
331
332         assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
333                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
334
335         // check follower has local shards
336         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
337                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
338
339         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
340                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH)));
341
342         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
343                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH)));
344
345         assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
346                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH)));
347
348         LOG.debug("Closing registrations");
349
350         reg1.close().toCompletableFuture().get();
351         reg2.close().toCompletableFuture().get();
352         reg3.close().toCompletableFuture().get();
353         reg4.close().toCompletableFuture().get();
354
355         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
356                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
357
358         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
359                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
360
361         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
362                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
363
364         waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
365                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
366
367         LOG.debug("All leader shards gone");
368
369         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
370                 ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
371
372         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
373                 ClusterUtils.getCleanShardName(TestModel.OUTER_CONTAINER_PATH));
374
375         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
376                 ClusterUtils.getCleanShardName(TestModel.INNER_LIST_PATH));
377
378         waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
379                 ClusterUtils.getCleanShardName(TestModel.JUNK_PATH));
380
381         LOG.debug("All follower shards gone");
382         LOG.info("testMultipleShardRegistrations ending");
383     }
384
385     @Test
386     @Ignore
387     public void testMultipleRegistrationsAtOnePrefix() throws Exception {
388         LOG.info("testMultipleRegistrationsAtOnePrefix starting");
389         initEmptyDatastores();
390
391         for (int i = 0; i < 10; i++) {
392             LOG.debug("Round {}", i);
393             final DistributedShardRegistration reg1 = waitOnAsyncTask(leaderShardFactory.createDistributedShard(
394                     TEST_ID, Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME)),
395                     DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
396
397             leaderTestKit.waitUntilLeader(leaderConfigDatastore.getActorContext(),
398                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
399
400             assertNotNull(findLocalShard(leaderConfigDatastore.getActorContext(),
401                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
402
403             assertNotNull(findLocalShard(followerConfigDatastore.getActorContext(),
404                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH)));
405
406
407             final Set<String> peers  = new HashSet<>();
408             IntegrationTestKit.verifyShardState(leaderConfigDatastore,
409                     ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()), onDemandShardState ->
410                             peers.addAll(onDemandShardState.getPeerAddresses().values()));
411             assertEquals(peers.size(), 1);
412
413             waitOnAsyncTask(reg1.close(), DistributedShardedDOMDataTree.SHARD_FUTURE_TIMEOUT_DURATION);
414
415             waitUntilShardIsDown(leaderConfigDatastore.getActorContext(),
416                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
417
418             waitUntilShardIsDown(followerConfigDatastore.getActorContext(),
419                     ClusterUtils.getCleanShardName(TestModel.TEST_PATH));
420         }
421
422         LOG.info("testMultipleRegistrationsAtOnePrefix ending");
423     }
424 }

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