Add Delayed Owner selection base on strategy

[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 e14ae7f5376e8e2f763b24dfe745bb82970d90c1..334e093d7f4babfafef4174275d7c32092decbe1 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
@@ -26,6 +26,7 @@ import static org.opendaylight.controller.cluster.datastore.entityownership.Enti
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 import akka.pattern.Patterns;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
@@ -42,9 +43,11 @@ import org.opendaylight.controller.cluster.datastore.entityownership.messages.Ca
 import org.opendaylight.controller.cluster.datastore.entityownership.messages.CandidateRemoved;
 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterCandidateLocal;
 import org.opendaylight.controller.cluster.datastore.entityownership.messages.RegisterListenerLocal;
+import org.opendaylight.controller.cluster.datastore.entityownership.messages.SelectOwner;
 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterCandidateLocal;
 import org.opendaylight.controller.cluster.datastore.entityownership.messages.UnregisterListenerLocal;
 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategy;
+import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.EntityOwnerSelectionStrategyWrapper;
 import org.opendaylight.controller.cluster.datastore.entityownership.selectionstrategy.FirstCandidateSelectionStrategy;
 import org.opendaylight.controller.cluster.datastore.identifiers.ShardIdentifier;
 import org.opendaylight.controller.cluster.datastore.messages.BatchedModifications;
@@ -70,16 +73,13 @@ import scala.concurrent.Future;
  * @author Thomas Pantelis
  */
 class EntityOwnershipShard extends Shard {
-
-    private static final EntityOwnerSelectionStrategy DEFAULT_ENTITY_OWNER_SELECTION_STRATEGY
-            = FirstCandidateSelectionStrategy.INSTANCE;
-
     private final String localMemberName;
     private final EntityOwnershipShardCommitCoordinator commitCoordinator;
     private final EntityOwnershipListenerSupport listenerSupport;
     private final Set<String> downPeerMemberNames = new HashSet<>();
     private final Map<String, String> peerIdToMemberNames = new HashMap<>();
-    private final Map<String, EntityOwnerSelectionStrategy> ownerSelectionStrategies = new HashMap<>();
+    private final Map<String, EntityOwnerSelectionStrategyWrapper> ownerSelectionStrategies = new HashMap<>();
+    private final EntityOwnerSelectionStrategyWrapper defaultEntityOwnerSelectionStrategy;
 
     private static DatastoreContext noPersistenceDatastoreContext(DatastoreContext datastoreContext) {
         return DatastoreContext.newBuilderFrom(datastoreContext).persistent(false).build();
@@ -90,6 +90,8 @@ class EntityOwnershipShard extends Shard {
         this.localMemberName = builder.localMemberName;
         this.commitCoordinator = new EntityOwnershipShardCommitCoordinator(builder.localMemberName, LOG);
         this.listenerSupport = new EntityOwnershipListenerSupport(getContext(), persistenceId());
+        this.defaultEntityOwnerSelectionStrategy =
+                createEntityOwnerSelectionStrategyWrapper(FirstCandidateSelectionStrategy.INSTANCE);
 
         for(String peerId: getRaftActorContext().getPeerIds()) {
             ShardIdentifier shardId = ShardIdentifier.builder().fromShardIdString(peerId).build();
@@ -113,7 +115,7 @@ class EntityOwnershipShard extends Shard {
     @Override
     public void onReceiveCommand(final Object message) throws Exception {
         if(message instanceof RegisterCandidateLocal) {
-            onRegisterCandidateLocal((RegisterCandidateLocal)message);
+            onRegisterCandidateLocal((RegisterCandidateLocal) message);
         } else if(message instanceof UnregisterCandidateLocal) {
             onUnregisterCandidateLocal((UnregisterCandidateLocal)message);
         } else if(message instanceof CandidateAdded){
@@ -124,15 +126,25 @@ class EntityOwnershipShard extends Shard {
             onPeerDown((PeerDown) message);
         } else if(message instanceof PeerUp) {
             onPeerUp((PeerUp) message);
-        } if(message instanceof RegisterListenerLocal) {
+        } else if(message instanceof RegisterListenerLocal) {
             onRegisterListenerLocal((RegisterListenerLocal)message);
-        } if(message instanceof UnregisterListenerLocal) {
-            onUnregisterListenerLocal((UnregisterListenerLocal)message);
+        } else if(message instanceof UnregisterListenerLocal) {
+            onUnregisterListenerLocal((UnregisterListenerLocal) message);
+        } else if(message instanceof SelectOwner) {
+            onSelectOwner((SelectOwner) message);
         } else if(!commitCoordinator.handleMessage(message, this)) {
             super.onReceiveCommand(message);
         }
     }
 
+    private void onSelectOwner(SelectOwner selectOwner) {
+        String currentOwner = getCurrentOwner(selectOwner.getEntityPath());
+        if(Strings.isNullOrEmpty(currentOwner)) {
+            writeNewOwner(selectOwner.getEntityPath(), newOwner(selectOwner.getAllCandidates(),
+                    selectOwner.getOwnerSelectionStrategy()));
+        }
+    }
+
     private void onRegisterCandidateLocal(RegisterCandidateLocal registerCandidate) {
         LOG.debug("{}: onRegisterCandidateLocal: {}", persistenceId(), registerCandidate);
 
@@ -250,8 +262,8 @@ class EntityOwnershipShard extends Shard {
         if(isLeader()) {
             String currentOwner = getCurrentOwner(message.getEntityPath());
             if(message.getRemovedCandidate().equals(currentOwner)){
-                writeNewOwner(message.getEntityPath(), newOwner(message.getRemainingCandidates(),
-                        getEntityOwnerElectionStrategy(message.getEntityPath())));
+                writeNewOwner(message.getEntityPath(),
+                        newOwner(message.getRemainingCandidates(), getEntityOwnerElectionStrategyWrapper(message.getEntityPath())));
             }
         } else {
             // We're not the leader. If the removed candidate is our local member then check if we actually
@@ -271,12 +283,12 @@ class EntityOwnershipShard extends Shard {
         }
     }
 
-    private EntityOwnerSelectionStrategy getEntityOwnerElectionStrategy(YangInstanceIdentifier entityPath) {
+    private EntityOwnerSelectionStrategyWrapper getEntityOwnerElectionStrategyWrapper(YangInstanceIdentifier entityPath) {
         String entityType = EntityOwnersModel.entityTypeFromEntityPath(entityPath);
-        EntityOwnerSelectionStrategy entityOwnerSelectionStrategy = ownerSelectionStrategies.get(entityType);
+        EntityOwnerSelectionStrategyWrapper entityOwnerSelectionStrategy = ownerSelectionStrategies.get(entityType);
 
         if(entityOwnerSelectionStrategy == null){
-            entityOwnerSelectionStrategy = DEFAULT_ENTITY_OWNER_SELECTION_STRATEGY;
+            entityOwnerSelectionStrategy = defaultEntityOwnerSelectionStrategy;
             ownerSelectionStrategies.put(entityType, entityOwnerSelectionStrategy);
         }
 
@@ -296,13 +308,11 @@ class EntityOwnershipShard extends Shard {
 
         String currentOwner = getCurrentOwner(message.getEntityPath());
         if(Strings.isNullOrEmpty(currentOwner)){
-            EntityOwnerSelectionStrategy entityOwnerSelectionStrategy
-                    = getEntityOwnerElectionStrategy(message.getEntityPath());
-            if(entityOwnerSelectionStrategy.selectionDelayInMillis() == 0L) {
-                writeNewOwner(message.getEntityPath(), newOwner(message.getAllCandidates(),
-                        entityOwnerSelectionStrategy));
+            EntityOwnerSelectionStrategyWrapper strategy = getEntityOwnerElectionStrategyWrapper(message.getEntityPath());
+            if(strategy.selectionDelayInMillis() == 0L) {
+                writeNewOwner(message.getEntityPath(), newOwner(message.getAllCandidates(), strategy));
             } else {
-                throw new UnsupportedOperationException("Delayed selection not implemented yet");
+                strategy.scheduleOwnerSelection(message.getEntityPath(), message.getAllCandidates());
             }
         }
     }
@@ -334,7 +344,7 @@ class EntityOwnershipShard extends Shard {
                         node(entityTypeNode.getIdentifier()).node(ENTITY_NODE_ID).node(entityNode.getIdentifier()).
                         node(ENTITY_OWNER_NODE_ID).build();
 
-                Object newOwner = newOwner(getCandidateNames(entityNode), getEntityOwnerElectionStrategy(entityPath));
+                Object newOwner = newOwner(getCandidateNames(entityNode), getEntityOwnerElectionStrategyWrapper(entityPath));
 
                 LOG.debug("{}: Found entity {}, writing new owner {}", persistenceId(), entityPath, newOwner);
 
@@ -351,9 +361,9 @@ class EntityOwnershipShard extends Shard {
         searchForEntities(new EntityWalker() {
             @Override
             public void onEntity(MapEntryNode entityTypeNode, MapEntryNode entityNode) {
-                if(hasCandidate(entityNode, owner)) {
+                if (hasCandidate(entityNode, owner)) {
                     YangInstanceIdentifier entityId =
-                            (YangInstanceIdentifier)entityNode.getIdentifier().getKeyValues().get(ENTITY_ID_QNAME);
+                            (YangInstanceIdentifier) entityNode.getIdentifier().getKeyValues().get(ENTITY_ID_QNAME);
                     YangInstanceIdentifier candidatePath = candidatePath(
                             entityTypeNode.getIdentifier().getKeyValues().get(ENTITY_TYPE_QNAME).toString(),
                             entityId, owner);
@@ -460,6 +470,22 @@ class EntityOwnershipShard extends Shard {
         void onEntity(MapEntryNode entityTypeNode, MapEntryNode entityNode);
     }
 
+    private EntityOwnerSelectionStrategyWrapper createEntityOwnerSelectionStrategyWrapper(EntityOwnerSelectionStrategy entityOwnerSelectionStrategy){
+        return new EntityOwnerSelectionStrategyWrapper(context().system().scheduler(), self(),
+                context().system().dispatcher(), entityOwnerSelectionStrategy);
+    }
+
+    @VisibleForTesting
+    void addEntityOwnerSelectionStrategy(String entityType, Class<? extends EntityOwnerSelectionStrategy> ownerSelectionStrategyClass){
+        try {
+            EntityOwnerSelectionStrategyWrapper strategy =
+                    createEntityOwnerSelectionStrategyWrapper(ownerSelectionStrategyClass.newInstance());
+            ownerSelectionStrategies.put(entityType, strategy);
+        } catch (InstantiationException | IllegalAccessException e) {
+            LOG.error("Exception occurred when adding election strategy", e);
+        }
+    }
+
     public static Builder newBuilder() {
         return new Builder();
     }