2 * Copyright (c) 2015 Brocade Communications 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
8 package org.opendaylight.controller.cluster.datastore.entityownership;
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertTrue;
12 import static org.mockito.AdditionalMatchers.or;
13 import static org.mockito.Mockito.atMost;
14 import static org.mockito.Mockito.doNothing;
15 import static org.mockito.Mockito.reset;
16 import static org.mockito.Mockito.timeout;
17 import static org.mockito.Mockito.verify;
18 import static org.opendaylight.controller.cluster.datastore.entityownership.AbstractEntityOwnershipTest.ownershipChange;
19 import static org.opendaylight.controller.cluster.datastore.entityownership.DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME;
20 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.CANDIDATE_NAME_NODE_ID;
21 import static org.opendaylight.controller.cluster.datastore.entityownership.EntityOwnersModel.entityPath;
22 import akka.actor.ActorSystem;
23 import akka.actor.Address;
24 import akka.actor.AddressFromURIString;
25 import akka.actor.Status.Failure;
26 import akka.actor.Status.Success;
27 import akka.cluster.Cluster;
28 import akka.testkit.JavaTestKit;
29 import com.google.common.base.Function;
30 import com.google.common.base.Optional;
31 import com.google.common.base.Stopwatch;
32 import com.google.common.util.concurrent.Uninterruptibles;
33 import com.typesafe.config.ConfigFactory;
34 import java.util.ArrayList;
35 import java.util.Arrays;
36 import java.util.Collection;
37 import java.util.concurrent.TimeUnit;
38 import org.junit.After;
39 import org.junit.Before;
40 import org.junit.Test;
41 import org.mockito.ArgumentCaptor;
42 import org.mockito.Mock;
43 import org.mockito.Mockito;
44 import org.mockito.MockitoAnnotations;
45 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
46 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
47 import org.opendaylight.controller.cluster.datastore.IntegrationTestKit;
48 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyConfig;
49 import org.opendaylight.controller.cluster.datastore.messages.AddShardReplica;
50 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
51 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
52 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
53 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidateRegistration;
54 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipChange;
55 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipListener;
56 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipState;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.clustering.entity.owners.rev150804.entity.owners.entity.type.entity.Candidate;
58 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
59 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
60 import org.opendaylight.yangtools.yang.data.api.schema.MapNode;
61 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
62 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
65 * End-to-end integration tests for the entity ownership functionality.
67 * @author Thomas Pantelis
69 public class DistributedEntityOwnershipIntegrationTest {
70 private static final Address MEMBER_1_ADDRESS = AddressFromURIString.parse("akka.tcp://cluster-test@127.0.0.1:2558");
71 private static final String MODULE_SHARDS_CONFIG = "module-shards-default.conf";
72 private static final String MODULE_SHARDS_MEMBER_1_CONFIG = "module-shards-default-member-1.conf";
73 private static final String ENTITY_TYPE1 = "entityType1";
74 private static final String ENTITY_TYPE2 = "entityType2";
75 private static final Entity ENTITY1 = new Entity(ENTITY_TYPE1, "entity1");
76 private static final Entity ENTITY1_2 = new Entity(ENTITY_TYPE2, "entity1");
77 private static final Entity ENTITY2 = new Entity(ENTITY_TYPE1, "entity2");
78 private static final Entity ENTITY3 = new Entity(ENTITY_TYPE1, "entity3");
79 private static final Entity ENTITY4 = new Entity(ENTITY_TYPE1, "entity4");
80 private static final SchemaContext SCHEMA_CONTEXT = SchemaContextHelper.entityOwners();
82 private ActorSystem leaderSystem;
83 private ActorSystem follower1System;
84 private ActorSystem follower2System;
86 private final DatastoreContext.Builder leaderDatastoreContextBuilder =
87 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(5);
89 private final DatastoreContext.Builder followerDatastoreContextBuilder =
90 DatastoreContext.newBuilder().shardHeartbeatIntervalInMillis(100).shardElectionTimeoutFactor(10000);
92 private DistributedDataStore leaderDistributedDataStore;
93 private DistributedDataStore follower1DistributedDataStore;
94 private DistributedDataStore follower2DistributedDataStore;
95 private DistributedEntityOwnershipService leaderEntityOwnershipService;
96 private DistributedEntityOwnershipService follower1EntityOwnershipService;
97 private DistributedEntityOwnershipService follower2EntityOwnershipService;
98 private IntegrationTestKit leaderTestKit;
99 private IntegrationTestKit follower1TestKit;
100 private IntegrationTestKit follower2TestKit;
103 private EntityOwnershipListener leaderMockListener;
106 private EntityOwnershipListener leaderMockListener2;
109 private EntityOwnershipListener follower1MockListener;
112 private EntityOwnershipListener follower2MockListener;
115 public void setUp() {
116 MockitoAnnotations.initMocks(this);
120 public void tearDown() {
121 if(leaderSystem != null) {
122 JavaTestKit.shutdownActorSystem(leaderSystem);
125 if(follower1System != null) {
126 JavaTestKit.shutdownActorSystem(follower1System);
129 if(follower2System != null) {
130 JavaTestKit.shutdownActorSystem(follower2System);
134 private void startAllSystems() {
136 startFollower1System();
137 startFollower2System();
140 private void startFollower2System() {
141 follower2System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member3"));
142 Cluster.get(follower2System).join(MEMBER_1_ADDRESS);
145 private void startFollower1System() {
146 follower1System = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member2"));
147 Cluster.get(follower1System).join(MEMBER_1_ADDRESS);
150 private void startLeaderSystem() {
151 leaderSystem = ActorSystem.create("cluster-test", ConfigFactory.load().getConfig("Member1"));
152 Cluster.get(leaderSystem).join(MEMBER_1_ADDRESS);
155 private void initDatastores(String type) {
156 initLeaderDatastore(type, MODULE_SHARDS_CONFIG);
158 initFollower1Datastore(type, MODULE_SHARDS_CONFIG);
160 follower2TestKit = new IntegrationTestKit(follower2System, followerDatastoreContextBuilder);
161 follower2DistributedDataStore = follower2TestKit.setupDistributedDataStore(
162 type, MODULE_SHARDS_CONFIG, false, SCHEMA_CONTEXT);
164 leaderDistributedDataStore.waitTillReady();
165 follower1DistributedDataStore.waitTillReady();
166 follower2DistributedDataStore.waitTillReady();
168 startLeaderService();
170 startFollower1Service();
172 follower2EntityOwnershipService = new DistributedEntityOwnershipService(follower2DistributedDataStore,
173 EntityOwnerSelectionStrategyConfig.newBuilder().build());
174 follower2EntityOwnershipService.start();
176 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
179 private void startFollower1Service() {
180 follower1EntityOwnershipService = new DistributedEntityOwnershipService(follower1DistributedDataStore,
181 EntityOwnerSelectionStrategyConfig.newBuilder().build());
182 follower1EntityOwnershipService.start();
185 private void startLeaderService() {
186 leaderEntityOwnershipService = new DistributedEntityOwnershipService(leaderDistributedDataStore,
187 EntityOwnerSelectionStrategyConfig.newBuilder().build());
188 leaderEntityOwnershipService.start();
191 private void initFollower1Datastore(String type, String moduleConfig) {
192 follower1TestKit = new IntegrationTestKit(follower1System, followerDatastoreContextBuilder);
193 follower1DistributedDataStore = follower1TestKit.setupDistributedDataStore(
194 type, moduleConfig, false, SCHEMA_CONTEXT);
197 private void initLeaderDatastore(String type, String moduleConfig) {
198 leaderTestKit = new IntegrationTestKit(leaderSystem, leaderDatastoreContextBuilder);
199 leaderDistributedDataStore = leaderTestKit.setupDistributedDataStore(
200 type, moduleConfig, false, SCHEMA_CONTEXT);
204 public void test() throws Exception {
206 initDatastores("test");
208 leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
209 leaderEntityOwnershipService.registerListener(ENTITY_TYPE2, leaderMockListener2);
210 follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
212 // Register leader candidate for entity1 and verify it becomes owner
214 leaderEntityOwnershipService.registerCandidate(ENTITY1);
215 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
216 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, false, true));
217 reset(leaderMockListener, follower1MockListener);
219 verifyGetOwnershipState(leaderEntityOwnershipService, ENTITY1, true, true);
220 verifyGetOwnershipState(follower1EntityOwnershipService, ENTITY1, false, true);
222 // Register leader candidate for entity1_2 (same id, different type) and verify it becomes owner
224 leaderEntityOwnershipService.registerCandidate(ENTITY1_2);
225 verify(leaderMockListener2, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1_2, false, true, true));
226 verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1_2));
227 reset(leaderMockListener2);
229 // Register follower1 candidate for entity1 and verify it gets added but doesn't become owner
231 follower1EntityOwnershipService.registerCandidate(ENTITY1);
232 verifyCandidates(leaderDistributedDataStore, ENTITY1, "member-1", "member-2");
233 verifyOwner(leaderDistributedDataStore, ENTITY1, "member-1");
234 verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
235 verify(follower1MockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
237 // Register follower1 candidate for entity2 and verify it becomes owner
239 follower1EntityOwnershipService.registerCandidate(ENTITY2);
240 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
241 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
242 reset(leaderMockListener, follower1MockListener);
244 // Register follower2 candidate for entity2 and verify it gets added but doesn't become owner
246 follower2EntityOwnershipService.registerListener(ENTITY_TYPE1, follower2MockListener);
247 follower2EntityOwnershipService.registerCandidate(ENTITY2);
248 verifyCandidates(leaderDistributedDataStore, ENTITY2, "member-2", "member-3");
249 verifyOwner(leaderDistributedDataStore, ENTITY2, "member-2");
251 // Unregister follower1 candidate for entity2 and verify follower2 becomes owner
253 follower1EntityOwnershipService.unregisterCandidate(ENTITY2);
254 verifyCandidates(leaderDistributedDataStore, ENTITY2, "member-3");
255 verifyOwner(leaderDistributedDataStore, ENTITY2, "member-3");
256 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, true, false, true));
257 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
259 // Depending on timing, follower2MockListener could get ownershipChanged with "false, false, true" if
260 // if the original ownership change with "member-2 is replicated to follower2 after the listener is
262 Uninterruptibles.sleepUninterruptibly(500, TimeUnit.MILLISECONDS);
263 verify(follower2MockListener, atMost(1)).ownershipChanged(ownershipChange(ENTITY2, false, false, true));
264 verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
266 // Register follower1 candidate for entity3 and verify it becomes owner
268 follower1EntityOwnershipService.registerCandidate(ENTITY3);
269 verifyOwner(leaderDistributedDataStore, ENTITY3, "member-2");
270 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, true, true));
271 verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, false, true));
272 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY3, false, false, true));
274 // Register follower2 candidate for entity4 and verify it becomes owner
276 follower2EntityOwnershipService.registerCandidate(ENTITY4);
277 verifyOwner(leaderDistributedDataStore, ENTITY4, "member-3");
278 verify(follower2MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, true, true));
279 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, false, true));
280 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY4, false, false, true));
281 reset(follower1MockListener, follower2MockListener);
283 // Register follower1 candidate for entity4 and verify it gets added but doesn't become owner
285 follower1EntityOwnershipService.registerCandidate(ENTITY4);
286 verifyCandidates(leaderDistributedDataStore, ENTITY4, "member-3", "member-2");
287 verifyOwner(leaderDistributedDataStore, ENTITY4, "member-3");
289 // Shutdown follower2 and verify it's owned entities (entity 2 & 4) get re-assigned
291 reset(leaderMockListener, follower1MockListener);
292 JavaTestKit.shutdownActorSystem(follower2System);
294 verify(follower1MockListener, timeout(15000).times(2)).ownershipChanged(or(ownershipChange(ENTITY4, false, true, true),
295 ownershipChange(ENTITY2, false, false, false)));
296 verify(leaderMockListener, timeout(15000).times(2)).ownershipChanged(or(ownershipChange(ENTITY4, false, false, true),
297 ownershipChange(ENTITY2, false, false, false)));
298 verifyOwner(leaderDistributedDataStore, ENTITY2, ""); // no other candidate
300 // Register leader candidate for entity2 and verify it becomes owner
302 leaderEntityOwnershipService.registerCandidate(ENTITY2);
303 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, true, true));
304 verifyOwner(leaderDistributedDataStore, ENTITY2, "member-1");
306 // Unregister leader candidate for entity2 and verify the owner is cleared
308 leaderEntityOwnershipService.unregisterCandidate(ENTITY2);
309 verifyOwner(leaderDistributedDataStore, ENTITY2, "");
310 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, true, false, false));
311 verify(follower1MockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY2, false, false, false));
315 * Reproduces bug <a href="https://bugs.opendaylight.org/show_bug.cgi?id=4554">4554</a>
317 * @throws CandidateAlreadyRegisteredException
320 public void testCloseCandidateRegistrationInQuickSuccession() throws CandidateAlreadyRegisteredException {
322 initDatastores("testCloseCandidateRegistrationInQuickSuccession");
324 leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
325 follower1EntityOwnershipService.registerListener(ENTITY_TYPE1, follower1MockListener);
326 follower2EntityOwnershipService.registerListener(ENTITY_TYPE1, follower2MockListener);
328 final EntityOwnershipCandidateRegistration candidate1 = leaderEntityOwnershipService.registerCandidate(ENTITY1);
329 final EntityOwnershipCandidateRegistration candidate2 = follower1EntityOwnershipService.registerCandidate(ENTITY1);
330 final EntityOwnershipCandidateRegistration candidate3 = follower2EntityOwnershipService.registerCandidate(ENTITY1);
332 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1, false, true, true));
334 Mockito.reset(leaderMockListener);
340 ArgumentCaptor<EntityOwnershipChange> leaderChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
341 ArgumentCaptor<EntityOwnershipChange> follower1ChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
342 ArgumentCaptor<EntityOwnershipChange> follower2ChangeCaptor = ArgumentCaptor.forClass(EntityOwnershipChange.class);
343 doNothing().when(leaderMockListener).ownershipChanged(leaderChangeCaptor.capture());
344 doNothing().when(follower1MockListener).ownershipChanged(follower1ChangeCaptor.capture());
345 doNothing().when(follower2MockListener).ownershipChanged(follower2ChangeCaptor.capture());
347 boolean passed = false;
348 for(int i=0;i<100;i++) {
349 Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
350 if(!leaderEntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
351 !follower1EntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
352 !follower2EntityOwnershipService.getOwnershipState(ENTITY1).get().hasOwner() &&
353 leaderChangeCaptor.getAllValues().size() > 0 && !leaderChangeCaptor.getValue().hasOwner() &&
354 leaderChangeCaptor.getAllValues().size() > 0 && !follower1ChangeCaptor.getValue().hasOwner() &&
355 leaderChangeCaptor.getAllValues().size() > 0 && !follower2ChangeCaptor.getValue().hasOwner()) {
361 assertTrue("No ownership change message was sent with hasOwner=false", passed);
365 * Tests bootstrapping the entity-ownership shard when there's no shards initially configured for local
366 * member. The entity-ownership shard is initially created as inactive (ie remains a follower), requiring
367 * an AddShardReplica request to join it to an existing leader.
370 public void testEntityOwnershipShardBootstrapping() throws Throwable {
372 startFollower1System();
373 String type = "testEntityOwnershipShardBootstrapping";
374 initLeaderDatastore(type, MODULE_SHARDS_MEMBER_1_CONFIG);
375 initFollower1Datastore(type, MODULE_SHARDS_MEMBER_1_CONFIG);
377 leaderDistributedDataStore.waitTillReady();
378 follower1DistributedDataStore.waitTillReady();
380 startLeaderService();
381 startFollower1Service();
383 leaderTestKit.waitUntilLeader(leaderDistributedDataStore.getActorContext(), ENTITY_OWNERSHIP_SHARD_NAME);
385 leaderEntityOwnershipService.registerListener(ENTITY_TYPE1, leaderMockListener);
387 // Register a candidate for follower1 - should get queued since follower1 has no leader
388 follower1EntityOwnershipService.registerCandidate(ENTITY1);
389 verify(leaderMockListener, timeout(300).never()).ownershipChanged(ownershipChange(ENTITY1));
391 // Add replica in follower1
392 AddShardReplica addReplica = new AddShardReplica(ENTITY_OWNERSHIP_SHARD_NAME);
393 follower1DistributedDataStore.getActorContext().getShardManager().tell(addReplica , follower1TestKit.getRef());
394 Object reply = follower1TestKit.expectMsgAnyClassOf(JavaTestKit.duration("5 sec"), Success.class, Failure.class);
395 if(reply instanceof Failure) {
396 throw ((Failure)reply).cause();
399 // The queued candidate registration should proceed
400 verify(leaderMockListener, timeout(5000)).ownershipChanged(ownershipChange(ENTITY1));
404 private static void verifyGetOwnershipState(DistributedEntityOwnershipService service, Entity entity,
405 boolean isOwner, boolean hasOwner) {
406 Optional<EntityOwnershipState> state = service.getOwnershipState(entity);
407 assertEquals("getOwnershipState present", true, state.isPresent());
408 assertEquals("isOwner", isOwner, state.get().isOwner());
409 assertEquals("hasOwner", hasOwner, state.get().hasOwner());
412 private static void verifyCandidates(DistributedDataStore dataStore, Entity entity, String... expCandidates) throws Exception {
413 AssertionError lastError = null;
414 Stopwatch sw = Stopwatch.createStarted();
415 while(sw.elapsed(TimeUnit.MILLISECONDS) <= 5000) {
416 Optional<NormalizedNode<?, ?>> possible = dataStore.newReadOnlyTransaction().read(
417 entityPath(entity.getType(), entity.getId()).node(Candidate.QNAME)).get(5, TimeUnit.SECONDS);
419 assertEquals("Candidates not found for " + entity, true, possible.isPresent());
420 Collection<String> actual = new ArrayList<>();
421 for(MapEntryNode candidate: ((MapNode)possible.get()).getValue()) {
422 actual.add(candidate.getChild(CANDIDATE_NAME_NODE_ID).get().getValue().toString());
425 assertEquals("Candidates for " + entity, Arrays.asList(expCandidates), actual);
427 } catch (AssertionError e) {
429 Uninterruptibles.sleepUninterruptibly(300, TimeUnit.MILLISECONDS);
436 private static void verifyOwner(final DistributedDataStore dataStore, Entity entity, String expOwner) {
437 AbstractEntityOwnershipTest.verifyOwner(expOwner, entity.getType(), entity.getId(),
438 new Function<YangInstanceIdentifier, NormalizedNode<?,?>>() {
440 public NormalizedNode<?, ?> apply(YangInstanceIdentifier path) {
442 return dataStore.newReadOnlyTransaction().read(path).get(5, TimeUnit.SECONDS).get();
443 } catch (Exception e) {