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.cluster.sharding.DistributedShardFactory.DistributedShardRegistration;
42 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
43 import org.opendaylight.controller.md.cluster.datastore.model.TestModel;
44 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
45 import org.opendaylight.mdsal.dom.api.DOMDataTreeCursorAwareTransaction;
46 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
47 import org.opendaylight.mdsal.dom.api.DOMDataTreeProducer;
48 import org.opendaylight.mdsal.dom.api.DOMDataTreeShardingConflictException;
49 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteCursor;
50 import org.opendaylight.mdsal.dom.broker.ShardedDOMDataTree;
51 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
52 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
53 import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableLeafNodeBuilder;
55 @Ignore("distributed-data is broken needs to be removed")
56 public class DistributedShardedDOMDataTreeTest extends AbstractTest {
58 private static final Address MEMBER_1_ADDRESS =
59 AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
61 private static final DOMDataTreeIdentifier TEST_ID =
62 new DOMDataTreeIdentifier(LogicalDatastoreType.CONFIGURATION, TestModel.TEST_PATH);
64 private ShardedDOMDataTree shardedDOMDataTree = new ShardedDOMDataTree();
66 private ActorSystem leaderSystem;
67 private ActorSystem followerSystem;
70 private final Builder leaderDatastoreContextBuilder =
71 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(2);
73 private final DatastoreContext.Builder followerDatastoreContextBuilder =
74 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5)
75 .customRaftPolicyImplementation(DisableElectionsRaftPolicy.class.getName());
77 private DistributedDataStore followerDistributedDataStore;
78 private DistributedDataStore leaderDistributedDataStore;
79 private IntegrationTestKit followerTestKit;
80 private IntegrationTestKit leaderTestKit;
82 private DistributedShardedDOMDataTree leaderShardFactory;
83 private DistributedShardedDOMDataTree followerShardFactory;
87 shardedDOMDataTree = new ShardedDOMDataTree();
89 leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
90 Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
92 followerSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
93 Cluster.get(followerSystem).join(MEMBER_1_ADDRESS);
97 public void tearDown() {
98 if (followerDistributedDataStore != null) {
99 leaderDistributedDataStore.close();
101 if (leaderDistributedDataStore != null) {
102 leaderDistributedDataStore.close();
105 JavaTestKit.shutdownActorSystem(leaderSystem);
106 JavaTestKit.shutdownActorSystem(followerSystem);
109 private void initEmptyDatastore(final String type) {
110 leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
112 leaderDistributedDataStore =
113 leaderTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full());
115 followerTestKit = new IntegrationTestKit(followerSystem, followerDatastoreContextBuilder);
116 followerDistributedDataStore =
117 followerTestKit.setupDistributedDataStoreWithoutConfig(type, SchemaContextHelper.full());
119 leaderShardFactory = new DistributedShardedDOMDataTree(leaderSystem,
120 Mockito.mock(DistributedDataStore.class),
121 leaderDistributedDataStore);
123 followerShardFactory = new DistributedShardedDOMDataTree(followerSystem,
124 Mockito.mock(DistributedDataStore.class),
125 followerDistributedDataStore);
129 public void testProducerRegistrations() throws Exception {
130 initEmptyDatastore("config");
132 final DistributedShardRegistration shardRegistration =
133 leaderShardFactory.createDistributedShard(TEST_ID,
134 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
136 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
137 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
139 final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
141 leaderShardManager.tell(
142 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef());
143 leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalShardFound.class);
145 IntegrationTestKit.findLocalShard(followerDistributedDataStore.getActorContext(),
146 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
148 leaderShardManager.tell(
149 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), leaderTestKit.getRef());
150 leaderTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), LocalPrimaryShardFound.class);
152 final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager();
153 followerShardManager.tell(
154 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), followerTestKit.getRef());
155 followerTestKit.expectMsgClass(JavaTestKit.duration("10 seconds"), RemotePrimaryShardFound.class);
157 final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
159 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
160 fail("Producer should be already registered on the other node");
161 } catch (final IllegalArgumentException e) {
162 assertTrue(e.getMessage().contains("is attached to producer"));
167 final DOMDataTreeProducer followerProducer =
168 followerShardFactory.createProducer(Collections.singleton(TEST_ID));
170 leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
171 fail("Producer should be already registered on the other node");
172 } catch (final IllegalArgumentException e) {
173 assertTrue(e.getMessage().contains("is attached to producer"));
176 followerProducer.close();
177 // try to create a shard on an already registered prefix on follower
179 followerShardFactory.createDistributedShard(TEST_ID,
180 Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
181 fail("This prefix already should have a shard registration that was forwarded from the other node");
182 } catch (final DOMDataTreeShardingConflictException e) {
183 assertTrue(e.getMessage().contains("is already occupied by shard"));
188 @Ignore("Needs some other stuff related to 5280")
189 public void testWriteIntoMultipleShards() throws Exception {
190 initEmptyDatastore("config");
192 final DistributedShardRegistration shardRegistration =
193 leaderShardFactory.createDistributedShard(
194 TEST_ID,Lists.newArrayList(AbstractTest.MEMBER_NAME, AbstractTest.MEMBER_2_NAME));
196 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(),
197 ClusterUtils.getCleanShardName(TEST_ID.getRootIdentifier()));
199 final ActorRef leaderShardManager = leaderDistributedDataStore.getActorContext().getShardManager();
201 new JavaTestKit(leaderSystem) {
203 leaderShardManager.tell(
204 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
205 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
207 final ActorRef followerShardManager = followerDistributedDataStore.getActorContext().getShardManager();
209 followerShardManager.tell(
210 new FindLocalShard(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
211 expectMsgClass(duration("5 seconds"), LocalShardFound.class);
213 leaderDistributedDataStore.getActorContext().getShardManager().tell(
214 new FindPrimary(ClusterUtils.getCleanShardName(TestModel.TEST_PATH), true), getRef());
215 expectMsgClass(duration("5 seconds"), LocalPrimaryShardFound.class);
219 final DOMDataTreeProducer producer = leaderShardFactory.createProducer(Collections.singleton(TEST_ID));
221 final DOMDataTreeCursorAwareTransaction tx = producer.createTransaction(true);
222 final DOMDataTreeWriteCursor cursor = tx.createCursor(TEST_ID);
223 Assert.assertNotNull(cursor);
224 final YangInstanceIdentifier nameId =
225 YangInstanceIdentifier.builder(TestModel.TEST_PATH).node(TestModel.NAME_QNAME).build();
226 cursor.write(nameId.getLastPathArgument(),
227 ImmutableLeafNodeBuilder.<String>create()
228 .withNodeIdentifier(new NodeIdentifier(TestModel.NAME_QNAME)).withValue("Test Value").build());