Add blueprint wiring to sal-distributed-datastore

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / entityownership / EntityOwnershipShard.java

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 fb7e9e4668b5cf7edf97996cf3629bf96fe150c8..678546e1670b3d5e6d3d4367c3c6a751f3cca910 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
@@ -113,7 +113,7 @@ class EntityOwnershipShard extends Shard {
     }
 
     @Override
-    public void onReceiveCommand(final Object message) throws Exception {
+    public void handleCommand(final Object message) {
         if(message instanceof RegisterCandidateLocal) {
             onRegisterCandidateLocal((RegisterCandidateLocal) message);
         } else if(message instanceof UnregisterCandidateLocal) {
@@ -133,7 +133,7 @@ class EntityOwnershipShard extends Shard {
         } else if(message instanceof SelectOwner) {
             onSelectOwner((SelectOwner) message);
         } else if(!commitCoordinator.handleMessage(message, this)) {
-            super.onReceiveCommand(message);
+            super.handleCommand(message);
         }
     }
 
@@ -143,7 +143,7 @@ class EntityOwnershipShard extends Shard {
         String currentOwner = getCurrentOwner(selectOwner.getEntityPath());
         if(Strings.isNullOrEmpty(currentOwner)) {
             String entityType = EntityOwnersModel.entityTypeFromEntityPath(selectOwner.getEntityPath());
-            writeNewOwner(selectOwner.getEntityPath(), newOwner(selectOwner.getAllCandidates(),
+            writeNewOwner(selectOwner.getEntityPath(), newOwner(currentOwner, selectOwner.getAllCandidates(),
                     entityOwnershipStatistics.byEntityType(entityType),
                     selectOwner.getOwnerSelectionStrategy()));
 
@@ -274,15 +274,17 @@ class EntityOwnershipShard extends Shard {
                 newLeader, isLeader);
 
         if(isLeader) {
-            // We were just elected leader. If the old leader is down, select new owners for the entities
-            // owned by the down leader.
 
-            String oldLeaderMemberName = peerIdToMemberNames.get(oldLeader);
+            // Clear all existing strategies so that they get re-created when we call createStrategy again
+            // This allows the strategies to be re-initialized with existing statistics maintained by
+            // EntityOwnershipStatistics
+            strategyConfig.clearStrategies();
 
-            LOG.debug("{}: oldLeaderMemberName: {}", persistenceId(), oldLeaderMemberName);
-
-            if(downPeerMemberNames.contains(oldLeaderMemberName)) {
-                removeCandidateFromEntities(oldLeaderMemberName);
+            // Remove the candidates for all members that are known to be down. In a cluster which has greater than
+            // 3 nodes it is possible for a some node beside the leader being down when the leadership transitions
+            // it makes sense to use this event to remove all the candidates for those downed nodes
+            for(String downPeerName : downPeerMemberNames){
+                removeCandidateFromEntities(downPeerName);
             }
         } else {
             // The leader changed - notify the coordinator to check if pending modifications need to be sent.
@@ -300,7 +302,7 @@ class EntityOwnershipShard extends Shard {
             if(message.getRemovedCandidate().equals(currentOwner) || message.getRemainingCandidates().size() == 0){
                 String entityType = EntityOwnersModel.entityTypeFromEntityPath(message.getEntityPath());
                 writeNewOwner(message.getEntityPath(),
-                        newOwner(message.getRemainingCandidates(), entityOwnershipStatistics.byEntityType(entityType),
+                        newOwner(currentOwner, message.getRemainingCandidates(), entityOwnershipStatistics.byEntityType(entityType),
                                 getEntityOwnerElectionStrategy(message.getEntityPath())));
             }
         } else {
@@ -323,7 +325,7 @@ class EntityOwnershipShard extends Shard {
 
     private EntityOwnerSelectionStrategy getEntityOwnerElectionStrategy(YangInstanceIdentifier entityPath) {
         final String entityType = EntityOwnersModel.entityTypeFromEntityPath(entityPath);
-        return strategyConfig.createStrategy(entityType);
+        return strategyConfig.createStrategy(entityType, entityOwnershipStatistics.byEntityType(entityType));
     }
 
     private void onCandidateAdded(CandidateAdded message) {
@@ -345,16 +347,17 @@ class EntityOwnershipShard extends Shard {
         // 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)){
+        LOG.debug("{}: Using strategy {} to select owner, currentOwner = {}", persistenceId(), strategy, currentOwner);
+
+        if(!message.getAllCandidates().contains(currentOwner)){
             if(strategy.getSelectionDelayInMillis() == 0L) {
-                writeNewOwner(message.getEntityPath(), newOwner(message.getAllCandidates(),
+                writeNewOwner(message.getEntityPath(), newOwner(currentOwner, 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(),
+                writeNewOwner(message.getEntityPath(), newOwner(currentOwner, message.getAllCandidates(),
                         entityOwnershipStatistics.byEntityType(entityType), strategy));
             } else {
                 scheduleOwnerSelection(message.getEntityPath(), message.getAllCandidates(), strategy);
@@ -464,12 +467,12 @@ class EntityOwnershipShard extends Shard {
         }
     }
 
-    private String newOwner(Collection<String> candidates, Map<String, Long> statistics, EntityOwnerSelectionStrategy ownerSelectionStrategy) {
+    private String newOwner(String currentOwner, Collection<String> candidates, Map<String, Long> statistics, EntityOwnerSelectionStrategy ownerSelectionStrategy) {
         Collection<String> viableCandidates = getViableCandidates(candidates);
         if(viableCandidates.size() == 0){
             return "";
         }
-        return ownerSelectionStrategy.newOwner(viableCandidates, statistics);
+        return ownerSelectionStrategy.newOwner(currentOwner, viableCandidates);
     }
 
     private Collection<String> getViableCandidates(Collection<String> candidates) {