opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/DistributedEntityOwnershipServiceTest.java

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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 package org.opendaylight.controller.cluster.datastore.entityownership;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12 import static org.junit.Assert.assertSame;
  13 import static org.junit.Assert.assertTrue;
  14 import static org.junit.Assert.fail;
  15 import static org.mockito.Mockito.mock;
  16 import akka.actor.ActorRef;
  17 import akka.actor.PoisonPill;
  18 import akka.actor.Props;
  19 import com.google.common.base.Optional;
  20 import com.google.common.base.Stopwatch;
  21 import com.google.common.collect.ImmutableMap;
  22 import com.google.common.util.concurrent.Uninterruptibles;
  23 import java.util.Collections;
  24 import java.util.List;
  25 import java.util.Map;
  26 import java.util.concurrent.CountDownLatch;
  27 import java.util.concurrent.TimeUnit;
  28 import java.util.concurrent.atomic.AtomicReference;
  29 import org.junit.After;
  30 import org.junit.Before;
  31 import org.junit.Test;
  32 import org.opendaylight.controller.cluster.datastore.DatastoreContext;
  33 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
  34 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
  35 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
  36 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
  37 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategy;
  38 import org.opendaylight.controller.cluster.datastore.shardstrategy.ShardStrategyFactory;
  39 import org.opendaylight.controller.cluster.datastore.utils.MockClusterWrapper;
  40 import org.opendaylight.controller.cluster.datastore.utils.MockConfiguration;
  41 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
  42 import org.opendaylight.controller.md.sal.common.api.clustering.CandidateAlreadyRegisteredException;
  43 import org.opendaylight.controller.md.sal.common.api.clustering.Entity;
  44 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidate;
  45 import org.opendaylight.controller.md.sal.common.api.clustering.EntityOwnershipCandidateRegistration;
  46 import org.opendaylight.controller.sal.core.spi.data.DOMStoreReadTransaction;
  47 import org.opendaylight.yangtools.yang.common.QName;
  48 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  49 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  50 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  51 import scala.concurrent.Await;
  52 import scala.concurrent.Future;
  53 import scala.concurrent.duration.Duration;
  54
  55 /**
  56  * Unit tests for DistributedEntityOwnershipService.
  57  *
  58  * @author Thomas Pantelis
  59  */
  60 public class DistributedEntityOwnershipServiceTest extends AbstractEntityOwnershipTest {
  61     static String ENTITY_TYPE = "test";
  62     static String ENTITY_TYPE2 = "test2";
  63     static int ID_COUNTER = 1;
  64     static final QName QNAME = QName.create("test", "2015-08-11", "foo");
  65
  66     private final String dataStoreType = "config" + ID_COUNTER++;
  67     private DistributedDataStore dataStore;
  68
  69     @Before
  70     public void setUp() {
  71         DatastoreContext datastoreContext = DatastoreContext.newBuilder().dataStoreType(dataStoreType).
  72                 shardInitializationTimeout(10, TimeUnit.SECONDS).build();
  73
  74         // FIXME - remove this MockConfiguration and use the production ConfigurationImpl class when the
  75         // DistributedEntityOwnershipService is changed to setup the ShardStrategy for the entity-owners module.
  76         MockConfiguration configuration = new MockConfiguration(Collections.<String, List<String>>emptyMap()) {
  77             @Override
  78             public Optional<String> getModuleNameFromNameSpace(String nameSpace) {
  79                 return Optional.of("entity-owners");
  80             }
  81
  82             @Override
  83             public Map<String, ShardStrategy> getModuleNameToShardStrategyMap() {
  84                 return ImmutableMap.<String, ShardStrategy>builder().put("entity-owners", new ShardStrategy() {
  85                     @Override
  86                     public String findShard(YangInstanceIdentifier path) {
  87                         return DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME;
  88                     }
  89                 }).build();
  90             }
  91         };
  92
  93         dataStore = new DistributedDataStore(getSystem(), new MockClusterWrapper(), configuration, datastoreContext );
  94
  95         dataStore.onGlobalContextUpdated(SchemaContextHelper.entityOwners());
  96
  97         ShardStrategyFactory.setConfiguration(configuration);
  98     }
  99
 100     @After
 101     public void tearDown() {
 102         dataStore.getActorContext().getShardManager().tell(PoisonPill.getInstance(), ActorRef.noSender());
 103     }
 104
 105     @Test
 106     public void testEntityOwnershipShardCreated() throws Exception {
 107         DistributedEntityOwnershipService service = new DistributedEntityOwnershipService(dataStore);
 108         service.start();
 109
 110         Future<ActorRef> future = dataStore.getActorContext().findLocalShardAsync(
 111                 DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME);
 112         ActorRef shardActor = Await.result(future, Duration.create(10, TimeUnit.SECONDS));
 113         assertNotNull(DistributedEntityOwnershipService.ENTITY_OWNERSHIP_SHARD_NAME + " not found", shardActor);
 114
 115         service.close();
 116     }
 117
 118     @Test
 119     public void testRegisterCandidate() throws Exception {
 120         final TestShardPropsCreator shardPropsCreator = new TestShardPropsCreator();
 121         DistributedEntityOwnershipService service = new DistributedEntityOwnershipService(dataStore) {
 122             @Override
 123             protected EntityOwnershipShardPropsCreator newShardPropsCreator() {
 124                 return shardPropsCreator;
 125             }
 126         };
 127
 128         service.start();
 129
 130         shardPropsCreator.expectShardMessage(RegisterCandidateLocal.class);
 131
 132         YangInstanceIdentifier entityId = YangInstanceIdentifier.of(QNAME);
 133         Entity entity = new Entity(ENTITY_TYPE, entityId);
 134         EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
 135
 136         EntityOwnershipCandidateRegistration reg = service.registerCandidate(entity, candidate);
 137
 138         verifyEntityOwnershipCandidateRegistration(entity, reg);
 139         verifyRegisterCandidateLocal(shardPropsCreator, entity, candidate);
 140         verifyEntityCandidate(readEntityOwners(service.getLocalEntityOwnershipShard()), ENTITY_TYPE, entityId,
 141                 dataStore.getActorContext().getCurrentMemberName());
 142
 143         // Register the same entity - should throw exception
 144
 145         EntityOwnershipCandidate candidate2 = mock(EntityOwnershipCandidate.class);
 146         try {
 147             service.registerCandidate(entity, candidate2);
 148             fail("Expected CandidateAlreadyRegisteredException");
 149         } catch(CandidateAlreadyRegisteredException e) {
 150             // expected
 151             assertSame("getCandidate", candidate, e.getRegisteredCandidate());
 152             assertEquals("getEntity", entity, e.getEntity());
 153         }
 154
 155         // Register a different entity - should succeed
 156
 157         Entity entity2 = new Entity(ENTITY_TYPE2, entityId);
 158         shardPropsCreator.expectShardMessage(RegisterCandidateLocal.class);
 159
 160         EntityOwnershipCandidateRegistration reg2 = service.registerCandidate(entity2, candidate);
 161
 162         verifyEntityOwnershipCandidateRegistration(entity2, reg2);
 163         verifyRegisterCandidateLocal(shardPropsCreator, entity2, candidate);
 164         verifyEntityCandidate(readEntityOwners(service.getLocalEntityOwnershipShard()), ENTITY_TYPE2, entityId,
 165                 dataStore.getActorContext().getCurrentMemberName());
 166
 167         service.close();
 168     }
 169
 170     @Test
 171     public void testCloseCandidateRegistration() throws Exception {
 172         final TestShardPropsCreator shardPropsCreator = new TestShardPropsCreator();
 173         DistributedEntityOwnershipService service = new DistributedEntityOwnershipService(dataStore) {
 174             @Override
 175             protected EntityOwnershipShardPropsCreator newShardPropsCreator() {
 176                 return shardPropsCreator;
 177             }
 178         };
 179
 180         service.start();
 181
 182         shardPropsCreator.expectShardMessage(RegisterCandidateLocal.class);
 183
 184         Entity entity = new Entity(ENTITY_TYPE, YangInstanceIdentifier.of(QNAME));
 185         EntityOwnershipCandidate candidate = mock(EntityOwnershipCandidate.class);
 186
 187         EntityOwnershipCandidateRegistration reg = service.registerCandidate(entity, candidate);
 188
 189         verifyEntityOwnershipCandidateRegistration(entity, reg);
 190         verifyRegisterCandidateLocal(shardPropsCreator, entity, candidate);
 191
 192         shardPropsCreator.expectShardMessage(UnregisterCandidateLocal.class);
 193
 194         reg.close();
 195
 196         UnregisterCandidateLocal unregCandidate = shardPropsCreator.waitForShardMessage();
 197         assertEquals("getEntity", entity, unregCandidate.getEntity());
 198
 199         // Re-register - should succeed.
 200
 201         shardPropsCreator.expectShardMessage(RegisterCandidateLocal.class);
 202
 203         service.registerCandidate(entity, candidate);
 204
 205         verifyRegisterCandidateLocal(shardPropsCreator, entity, candidate);
 206
 207         service.close();
 208     }
 209
 210     @Test
 211     public void testRegisterListener() {
 212     }
 213
 214     private NormalizedNode<?, ?> readEntityOwners(ActorRef shard) throws Exception {
 215         Stopwatch sw = Stopwatch.createStarted();
 216         while(sw.elapsed(TimeUnit.MILLISECONDS) <= 5000) {
 217             DOMStoreReadTransaction readTx = dataStore.newReadOnlyTransaction();
 218             Optional<NormalizedNode<?, ?>> optional = readTx.read(EntityOwnershipShard.ENTITY_OWNERS_PATH).
 219                     checkedGet(5, TimeUnit.SECONDS);
 220             if(optional.isPresent()) {
 221                 return optional.get();
 222             }
 223
 224             Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
 225         }
 226
 227         return null;
 228     }
 229
 230     private void verifyRegisterCandidateLocal(final TestShardPropsCreator shardPropsCreator, Entity entity,
 231             EntityOwnershipCandidate candidate) {
 232         RegisterCandidateLocal regCandidate = shardPropsCreator.waitForShardMessage();
 233         assertSame("getCandidate", candidate, regCandidate.getCandidate());
 234         assertEquals("getEntity", entity, regCandidate.getEntity());
 235     }
 236
 237     private void verifyEntityOwnershipCandidateRegistration(Entity entity, EntityOwnershipCandidateRegistration reg) {
 238         assertNotNull("EntityOwnershipCandidateRegistration null", reg);
 239         assertEquals("getEntity", entity, reg.getEntity());
 240     }
 241
 242     static class TestShardPropsCreator extends EntityOwnershipShardPropsCreator {
 243         TestShardPropsCreator() {
 244             super("member-1");
 245         }
 246
 247         private final AtomicReference<CountDownLatch> messageReceived = new AtomicReference<>();
 248         private final AtomicReference<Object> receivedMessage = new AtomicReference<>();
 249         private final AtomicReference<Class<?>> messageClass = new AtomicReference<>();
 250
 251         @Override
 252         public Props newProps(ShardIdentifier shardId, Map<String, String> peerAddresses,
 253                 DatastoreContext datastoreContext, SchemaContext schemaContext) {
 254             return Props.create(TestEntityOwnershipShard.class, shardId, peerAddresses, datastoreContext,
 255                     schemaContext, "member-1", messageClass, messageReceived, receivedMessage);
 256         }
 257
 258         @SuppressWarnings("unchecked")
 259         <T> T waitForShardMessage() {
 260             assertTrue("Message " + messageClass.get().getSimpleName() + " was not received",
 261                     Uninterruptibles.awaitUninterruptibly(messageReceived.get(), 5, TimeUnit.SECONDS));
 262             assertEquals("Message type", messageClass.get(), receivedMessage.get().getClass());
 263             return (T) receivedMessage.get();
 264         }
 265
 266         void expectShardMessage(Class<?> ofType) {
 267             messageReceived.set(new CountDownLatch(1));
 268             receivedMessage.set(null);
 269             messageClass.set(ofType);
 270         }
 271     }
 272
 273     static class TestEntityOwnershipShard extends EntityOwnershipShard {
 274         private final AtomicReference<CountDownLatch> messageReceived;
 275         private final AtomicReference<Object> receivedMessage;
 276         private final AtomicReference<Class<?>> messageClass;
 277
 278         protected TestEntityOwnershipShard(ShardIdentifier name, Map<String, String> peerAddresses,
 279                 DatastoreContext datastoreContext, SchemaContext schemaContext, String localMemberName,
 280                 AtomicReference<Class<?>> messageClass, AtomicReference<CountDownLatch> messageReceived,
 281                 AtomicReference<Object> receivedMessage) {
 282             super(name, peerAddresses, datastoreContext, schemaContext, localMemberName);
 283             this.messageClass = messageClass;
 284             this.messageReceived = messageReceived;
 285             this.receivedMessage = receivedMessage;
 286         }
 287
 288         @Override
 289         public void onReceiveCommand(final Object message) throws Exception {
 290             try {
 291                 super.onReceiveCommand(message);
 292             } finally {
 293                 Class<?> expMsgClass = messageClass.get();
 294                 if(expMsgClass != null && expMsgClass.equals(message.getClass())) {
 295                     receivedMessage.set(message);
 296                     messageReceived.get().countDown();
 297                 }
 298             }
 299         }
 300     }
 301 }