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

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