Choose owner when all candidate registrations received.

In the Delayed Owner Selection Strategy we should not wait for
the timeout to occur when we have received the candidate
registrations for all the candidates possible in the system.

Change-Id: Ifcd1f376b050baf2e422e00bd4a93a4d9d3d6c45
Signed-off-by: Moiz Raja <moraja@cisco.com>

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShard.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShard.java
index 89b7518a0326f5d1ae7044bf938e9ea7a5b20da1..3a031fc6a4313b6a0177e2c8c66287ec00912676 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShard.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShard.java
@@ -325,13 +325,23 @@ class EntityOwnershipShard extends Shard {
         // remove it from the downPeerMemberNames.
         downPeerMemberNames.remove(message.getNewCandidate());
 
-        String currentOwner = getCurrentOwner(message.getEntityPath());
-        EntityOwnerSelectionStrategy strategy = getEntityOwnerElectionStrategy(message.getEntityPath());
+        final String currentOwner = getCurrentOwner(message.getEntityPath());
+        final EntityOwnerSelectionStrategy strategy = getEntityOwnerElectionStrategy(message.getEntityPath());
+        final String entityType = EntityOwnersModel.entityTypeFromEntityPath(message.getEntityPath());
+
+        // Available members is all the known peers - the number of peers that are down + self
+        // So if there are 2 peers and 1 is down then availableMembers will be 2
+        final int availableMembers = (peerIdToMemberNames.size() - downPeerMemberNames.size()) + 1;
 
         LOG.debug("{}: Using strategy {} to select owner", persistenceId(), strategy);
         if(Strings.isNullOrEmpty(currentOwner)){
             if(strategy.getSelectionDelayInMillis() == 0L) {
-                String entityType = EntityOwnersModel.entityTypeFromEntityPath(message.getEntityPath());
+                writeNewOwner(message.getEntityPath(), newOwner(message.getAllCandidates(),
+                        entityOwnershipStatistics.byEntityType(entityType), strategy));
+            } else if(message.getAllCandidates().size() == availableMembers) {
+                LOG.debug("{}: Received the maximum candidates requests : {} writing new owner",
+                        persistenceId(), availableMembers);
+                cancelOwnerSelectionTask(message.getEntityPath());
                 writeNewOwner(message.getEntityPath(), newOwner(message.getAllCandidates(),
                         entityOwnershipStatistics.byEntityType(entityType), strategy));
             } else {
@@ -443,14 +453,11 @@ class EntityOwnershipShard extends Shard {
      */
     public void scheduleOwnerSelection(YangInstanceIdentifier entityPath, Collection<String> allCandidates,
                                        EntityOwnerSelectionStrategy strategy){
-        Cancellable lastScheduledTask = entityToScheduledOwnershipTask.get(entityPath);
-        if(lastScheduledTask != null && !lastScheduledTask.isCancelled()){
-            lastScheduledTask.cancel();
-        }
+        cancelOwnerSelectionTask(entityPath);
 
         LOG.debug("{}: Scheduling owner selection after {} ms", persistenceId(), strategy.getSelectionDelayInMillis());
 
-        lastScheduledTask = context().system().scheduler().scheduleOnce(
+        final Cancellable lastScheduledTask = context().system().scheduler().scheduleOnce(
                 FiniteDuration.apply(strategy.getSelectionDelayInMillis(), TimeUnit.MILLISECONDS)
                 , self(), new SelectOwner(entityPath, allCandidates, strategy)
                 , context().system().dispatcher(), self());
@@ -458,6 +465,13 @@ class EntityOwnershipShard extends Shard {
         entityToScheduledOwnershipTask.put(entityPath, lastScheduledTask);
     }
 
+    private void cancelOwnerSelectionTask(YangInstanceIdentifier entityPath){
+        final Cancellable lastScheduledTask = entityToScheduledOwnershipTask.get(entityPath);
+        if(lastScheduledTask != null && !lastScheduledTask.isCancelled()){
+            lastScheduledTask.cancel();
+        }
+    }
+
     private String newOwner(Collection<String> candidates, Map<String, Long> statistics, EntityOwnerSelectionStrategy ownerSelectionStrategy) {
         Collection<String> viableCandidates = getViableCandidates(candidates);
         if(viableCandidates.size() == 0){

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/selectionstrategy/LeastLoadedCandidateSelectionStrategy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/selectionstrategy/LeastLoadedCandidateSelectionStrategy.java
index 9ebf21cdcec1be10479cc5674219455fae42502b..b754046974b2fa48bc641f2c16860e69ec61db43 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/selectionstrategy/LeastLoadedCandidateSelectionStrategy.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/entityownership/selectionstrategy/LeastLoadedCandidateSelectionStrategy.java
@@ -8,14 +8,22 @@
 
 package org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy;
 
+import com.google.common.base.MoreObjects;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The LeastLoadedCandidateSelectionStrategy assigns ownership for an entity to the candidate which owns the least
  * number of entities.
  */
 public class LeastLoadedCandidateSelectionStrategy extends AbstractEntityOwnerSelectionStrategy {
+    private static final Logger LOG = LoggerFactory.getLogger(LeastLoadedCandidateSelectionStrategy.class);
+
+    private Map<String, Long> localStatistics = new HashMap<>();
+
     protected LeastLoadedCandidateSelectionStrategy(long selectionDelayInMillis) {
         super(selectionDelayInMillis);
     }
@@ -26,16 +34,28 @@ public class LeastLoadedCandidateSelectionStrategy extends AbstractEntityOwnerSe
         long leastLoadedCount = Long.MAX_VALUE;
 
         for(String candidateName : viableCandidates){
-            Long val = statistics.get(candidateName);
-            if(val != null && val < leastLoadedCount){
+            long val = MoreObjects.firstNonNull(statistics.get(candidateName), 0L);
+            long localVal = MoreObjects.firstNonNull(localStatistics.get(candidateName), 0L);
+            if(val < localVal){
+                LOG.debug("Local statistic higher - Candidate : {}, local statistic : {}, provided statistic : {}",
+                        candidateName, localVal, val);
+                val = localVal;
+            } else {
+                LOG.debug("Provided statistic higher - Candidate : {}, local statistic : {}, provided statistic : {}",
+                        candidateName, localVal, val);
+                localStatistics.put(candidateName, val);
+            }
+            if(val < leastLoadedCount){
                 leastLoadedCount = val;
                 leastLoadedCandidate = candidateName;
             }
         }
 
         if(leastLoadedCandidate == null){
-            return viableCandidates.iterator().next();
+            leastLoadedCandidate = viableCandidates.iterator().next();
         }
+
+        localStatistics.put(leastLoadedCandidate, leastLoadedCount + 1);
         return leastLoadedCandidate;
     }
 }

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java
index 4dd64bd95dee2ab93d790449d992b839c8dfded2..1cfee5f6b96989f6985210cce5ccbd06ea973008 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/entityownership/EntityOwnershipShardTest.java
@@ -809,6 +809,11 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
                 LOCAL_MEMBER_NAME, strategyConfig);
     }
 
+    private Props newShardProps(EntityOwnerSelectionStrategyConfig strategyConfig, Map<String, String> peers) {
+        return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, strategyConfig);
+    }
+
+
     private Props newShardProps(Map<String,String> peers) {
         return newShardProps(newShardId(LOCAL_MEMBER_NAME), peers, LOCAL_MEMBER_NAME, EntityOwnerSelectionStrategyConfig.newBuilder().build());
     }
@@ -882,11 +887,19 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
 
 
     @Test
-    public void testDelayedEntityOwnerSelection() throws Exception {
+    public void testDelayedEntityOwnerSelectionWhenMaxPeerRequestsReceived() throws Exception {
         ShardTestKit kit = new ShardTestKit(getSystem());
         EntityOwnerSelectionStrategyConfig.Builder builder
                 = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
-        TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build()));
+
+        String peerId = newShardId("follower").toString();
+        TestActorRef<MockFollower> peer = actorFactory.createTestActor(Props.create(MockFollower.class, peerId, false).
+                withDispatcher(Dispatchers.DefaultDispatcherId()), peerId);
+
+        peer.underlyingActor().grantVote = true;
+
+        TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
+                ImmutableMap.of(peerId.toString(), peer.path().toString())));
         kit.waitUntilLeader(shard);
 
         Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
@@ -894,9 +907,51 @@ public class EntityOwnershipShardTest extends AbstractEntityOwnershipTest {
 
         // Add a remote candidate
 
-        String remoteMemberName1 = "remoteMember1";
+        String remoteMemberName1 = "follower";
         writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 
+        // Register local
+
+        shard.tell(new RegisterCandidateLocal(entity), kit.getRef());
+        kit.expectMsgClass(SuccessReply.class);
+
+        // Verify the local candidate becomes owner
+
+        verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, remoteMemberName1);
+        verifyCommittedEntityCandidate(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
+        verifyOwner(shard, ENTITY_TYPE, ENTITY_ID1, LOCAL_MEMBER_NAME);
+    }
+
+    @Test
+    public void testDelayedEntityOwnerSelection() throws Exception {
+        ShardTestKit kit = new ShardTestKit(getSystem());
+        EntityOwnerSelectionStrategyConfig.Builder builder
+                = EntityOwnerSelectionStrategyConfig.newBuilder().addStrategy(ENTITY_TYPE, LastCandidateSelectionStrategy.class, 500);
+
+        String follower1Id = newShardId("follower1").toString();
+        TestActorRef<MockFollower> follower1 = actorFactory.createTestActor(Props.create(MockFollower.class, follower1Id, false).
+                withDispatcher(Dispatchers.DefaultDispatcherId()), follower1Id);
+
+        follower1.underlyingActor().grantVote = true;
+
+        String follower2Id = newShardId("follower").toString();
+        TestActorRef<MockFollower> follower2 = actorFactory.createTestActor(Props.create(MockFollower.class, follower2Id, false).
+                withDispatcher(Dispatchers.DefaultDispatcherId()), follower2Id);
+
+        follower2.underlyingActor().grantVote = true;
+
+
+        TestActorRef<EntityOwnershipShard> shard = actorFactory.createTestActor(newShardProps(builder.build(),
+                ImmutableMap.of(follower1Id.toString(), follower2.path().toString(), follower2Id.toString(), follower2.path().toString())));
+        kit.waitUntilLeader(shard);
+
+        Entity entity = new Entity(ENTITY_TYPE, ENTITY_ID1);
+        ShardDataTree shardDataTree = shard.underlyingActor().getDataStore();
+
+        // Add a remote candidate
+
+        String remoteMemberName1 = "follower";
+        writeNode(ENTITY_OWNERS_PATH, entityOwnersWithCandidate(ENTITY_TYPE, ENTITY_ID1, remoteMemberName1), shardDataTree);
 
         // Register local