2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.sharding;
11 import static org.junit.Assert.assertTrue;
12 import static org.junit.Assert.fail;
14 import akka.actor.ActorRef;
15 import akka.actor.ActorSystem;
16 import akka.actor.Address;
17 import akka.actor.AddressFromURIString;
18 import akka.cluster.Cluster;
19 import akka.testkit.JavaTestKit;
20 import com.google.common.collect.Lists;
21 import com.typesafe.config.ConfigFactory;
22 import java.util.Collections;
23 import org.junit.After;
24 import org.junit.Assert;
25 import org.junit.Before;
26 import org.junit.Ignore;
27 import org.junit.Test;
28 import org.mockito.Mockito;
29 import org.opendaylight.controller.cluster.datastore.AbstractTest;
30 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
31 import org.opendaylight.controller.cluster.datastore.DatastoreContext.Builder;
32 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
33 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
34 import org.opendaylight.controller.cluster.datastore.messages.FindLocalShard;
35 import org.opendaylight.controller.cluster.datastore.messages.FindPrimary;
36 import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardFound;
37 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
38 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
39 import org.opendaylight.controller.cluster.datastore.utils.ClusterUtils;
40 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
41 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
42 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
43 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
44 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
45 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
46 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
47 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
48 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
49 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
50 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
51 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
53 public class DistributedShardedDOMDataTreeTest extends AbstractTest {
55 private static final Address MEMBER_1_ADDRESS =
56 AddressFromURIString.parse("akka://cluster-test@127.0.0.1:2558");
58 private static final DOMDataTreeIdentifier TEST_ID =
59 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
61 private ActorSystem leaderSystem;
62 private ActorSystem followerSystem;
65 private final Builder leaderDatastoreContextBuilder =
66 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
68 private final DatastoreContext.Builder followerDatastoreContextBuilder =
69 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5)
70 .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
72 private DistributedDataStore followerDistributedDataStore;
73 private DistributedDataStore leaderDistributedDataStore;
74 private IntegrationTestKit followerTestKit;
75 private IntegrationTestKit leaderTestKit;
77 private DistributedShardedDOMDataTree leaderShardFactory;
78 private DistributedShardedDOMDataTree followerShardFactory;
82 leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
83 Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
85 followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
86 Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
90 public void tearDown() {
91 if (followerDistributedDataStore != null) {
92 leaderDistributedDataStore.close();
94 if (leaderDistributedDataStore != null) {
95 leaderDistributedDataStore.close();
98 JavaTestKit.shutdownActorSystem(leaderSystem);
99 JavaTestKit.shutdownActorSystem(followerSystem);
102 private void initEmptyDatastore(final String type) {
103 leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
105 leaderDistributedDataStore =
106 leaderTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full());
108 followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
109 followerDistributedDataStore =
110 followerTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full());
112 leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystem,
113 Mockito.mock(DistributedDataStore.class),
114 leaderDistributedDataStore);
116 followerShardFactory = new DistributedShardedDOMDataTree(followerSystem,
117 Mockito.mock(DistributedDataStore.class),
118 followerDistributedDataStore);
122 public void testProducerRegistrations() throws Exception {
123 initEmptyDatastore("config");
125 leaderShardFactory.createDistributedShard(TEST_ID,
126 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
128 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
129 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
131 final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
133 leaderShardManager.tell(
134 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef());
135 leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalShardFound.class);
137 IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorContext(),
138 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
140 leaderShardManager.tell(
141 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef());
142 leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalPrimaryShardFound.class);
144 final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager();
145 followerShardManager.tell(
146 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), followerTestKit.getRef());
147 followerTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), RemotePrimaryShardFound.class);
149 final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
151 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
152 fail("Producer should be already registered on the other node");
153 } catch (final IllegalArgumentException e) {
154 assertTrue(e.getMessage().contains("is attached to producer"));
159 final DOMDataTreeProducer followerProducer =
160 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
162 leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
163 fail("Producer should be already registered on the other node");
164 } catch (final IllegalArgumentException e) {
165 assertTrue(e.getMessage().contains("is attached to producer"));
168 followerProducer.close();
169 // try to create a shard on an already registered prefix on follower
171 followerShardFactory.createDistributedShard(TEST_ID,
172 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
173 fail("This prefix already should have a shard registration that was forwarded from the other node");
174 } catch (final DOMDataTreeShardingConflictException e) {
175 assertTrue(e.getMessage().contains("is already occupied by shard"));
180 @Ignore("Needs some other stuff related to 5280")
181 public void testWriteIntoMultipleShards() throws Exception {
182 initEmptyDatastore("config");
184 leaderShardFactory.createDistributedShard(TEST_ID,
185 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
187 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
188 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
190 final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
192 new JavaTestKit(leaderSystem) {
194 leaderShardManager.tell(
195 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
196 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
198 final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager();
200 followerShardManager.tell(
201 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
202 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
204 leaderDistributedDataStore.getActorContext().getShardManager().tell(
205 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
206 expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
210 final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
212 final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
213 final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
214 Assert.assertNotNull(cursor);
215 final YangInstanceIdentifier nameId =
216 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
217 cursor.write(nameId.getLastPathArgument(),
218 ImmutableLeafNodeBuilder.<String>create()
219 .withNodeIdentifier(new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build());