Bug 4094: Fix DCNs on initial registration

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

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
index 99d66d515f5b77e7afc36b570a23491878c59df0..60894f5bc9723059ef7441aa79357084a00f2345 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ShardManager.java
@@ -39,6 +39,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedPersistentActorWithMetering;
 import org.opendaylight.controller.cluster.datastore.config.Configuration;
+import org.opendaylight.controller.cluster.datastore.config.ModuleShardConfiguration;
 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
 import org.opendaylight.controller.cluster.datastore.exceptions.NotInitializedException;
 import org.opendaylight.controller.cluster.datastore.exceptions.PrimaryNotFoundException;
@@ -55,6 +56,8 @@ import org.opendaylight.controller.cluster.datastore.messages.LocalPrimaryShardF
 import org.opendaylight.controller.cluster.datastore.messages.LocalShardFound;
 import org.opendaylight.controller.cluster.datastore.messages.LocalShardNotFound;
 import org.opendaylight.controller.cluster.datastore.messages.PeerAddressResolved;
+import org.opendaylight.controller.cluster.datastore.messages.PeerDown;
+import org.opendaylight.controller.cluster.datastore.messages.PeerUp;
 import org.opendaylight.controller.cluster.datastore.messages.RemoteFindPrimary;
 import org.opendaylight.controller.cluster.datastore.messages.RemotePrimaryShardFound;
 import org.opendaylight.controller.cluster.datastore.messages.ShardLeaderStateChanged;
@@ -176,6 +179,8 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
             onActorInitialized(message);
         } else if (message instanceof ClusterEvent.MemberUp){
             memberUp((ClusterEvent.MemberUp) message);
+        } else if (message instanceof ClusterEvent.MemberExited){
+            memberExited((ClusterEvent.MemberExited) message);
         } else if(message instanceof ClusterEvent.MemberRemoved) {
             memberRemoved((ClusterEvent.MemberRemoved) message);
         } else if(message instanceof ClusterEvent.UnreachableMember) {
@@ -205,24 +210,29 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
     private void onCreateShard(CreateShard createShard) {
         Object reply;
         try {
-            if(localShards.containsKey(createShard.getShardName())) {
+            ModuleShardConfiguration moduleShardConfig = createShard.getModuleShardConfig();
+            if(localShards.containsKey(moduleShardConfig.getShardName())) {
                 throw new IllegalStateException(String.format("Shard with name %s already exists",
-                        createShard.getShardName()));
+                        moduleShardConfig.getShardName()));
             }
 
-            ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), createShard.getShardName());
-            Map<String, String> peerAddresses = getPeerAddresses(createShard.getShardName(), createShard.getMemberNames());
+            configuration.addModuleShardConfiguration(moduleShardConfig);
 
-            LOG.debug("onCreateShard: shardId: {}, peerAddresses: {}", shardId, peerAddresses);
+            ShardIdentifier shardId = getShardIdentifier(cluster.getCurrentMemberName(), moduleShardConfig.getShardName());
+            Map<String, String> peerAddresses = getPeerAddresses(moduleShardConfig.getShardName()/*,
+                    moduleShardConfig.getShardMemberNames()*/);
+
+            LOG.debug("onCreateShard: shardId: {}, memberNames: {}. peerAddresses: {}", shardId,
+                    moduleShardConfig.getShardMemberNames(), peerAddresses);
 
             DatastoreContext shardDatastoreContext = createShard.getDatastoreContext();
             if(shardDatastoreContext == null) {
                 shardDatastoreContext = datastoreContext;
             }
 
-            ShardInformation info = new ShardInformation(createShard.getShardName(), shardId, peerAddresses,
+            ShardInformation info = new ShardInformation(moduleShardConfig.getShardName(), shardId, peerAddresses,
                     shardDatastoreContext, createShard.getShardPropsCreator());
-            localShards.put(createShard.getShardName(), info);
+            localShards.put(info.getShardName(), info);
 
             mBean.addLocalShard(shardId.toString());
 
@@ -464,7 +474,24 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
         LOG.debug("{}: Received MemberRemoved: memberName: {}, address: {}", persistenceId(), memberName,
                 message.member().address());
 
-        memberNameToAddress.remove(message.member().roles().head());
+        memberNameToAddress.remove(memberName);
+
+        for(ShardInformation info : localShards.values()){
+            info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
+        }
+    }
+
+    private void memberExited(ClusterEvent.MemberExited message) {
+        String memberName = message.member().roles().head();
+
+        LOG.debug("{}: Received MemberExited: memberName: {}, address: {}", persistenceId(), memberName,
+                message.member().address());
+
+        memberNameToAddress.remove(memberName);
+
+        for(ShardInformation info : localShards.values()){
+            info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
+        }
     }
 
     private void memberUp(ClusterEvent.MemberUp message) {
@@ -477,8 +504,10 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
 
         for(ShardInformation info : localShards.values()){
             String shardName = info.getShardName();
-            info.updatePeerAddress(getShardIdentifier(memberName, shardName).toString(),
-                getShardActorPath(shardName, memberName), getSelf());
+            String peerId = getShardIdentifier(memberName, shardName).toString();
+            info.updatePeerAddress(peerId, getShardActorPath(shardName, memberName), getSelf());
+
+            info.peerUp(memberName, peerId, getSelf());
         }
 
         checkReady();
@@ -507,6 +536,8 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
 
                 primaryShardInfoCache.remove(info.getShardName());
             }
+
+            info.peerDown(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
         }
     }
 
@@ -517,6 +548,8 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
                 LOG.debug("Marking Leader {} as available.", leaderId);
                 info.setLeaderAvailable(true);
             }
+
+            info.peerUp(memberName, getShardIdentifier(memberName, info.getShardName()).toString(), getSelf());
         }
     }
 
@@ -657,8 +690,7 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
      */
     private void createLocalShards() {
         String memberName = this.cluster.getCurrentMemberName();
-        List<String> memberShardNames =
-            this.configuration.getMemberShardNames(memberName);
+        Collection<String> memberShardNames = this.configuration.getMemberShardNames(memberName);
 
         ShardPropsCreator shardPropsCreator = new DefaultShardPropsCreator();
         List<String> localShardActorNames = new ArrayList<>();
@@ -680,14 +712,9 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
      * Given the name of the shard find the addresses of all it's peers
      *
      * @param shardName
-     * @return
      */
     private Map<String, String> getPeerAddresses(String shardName) {
-        return getPeerAddresses(shardName, configuration.getMembersFromShardName(shardName));
-    }
-
-    private Map<String, String> getPeerAddresses(String shardName, Collection<String> members) {
-
+        Collection<String> members = configuration.getMembersFromShardName(shardName);
         Map<String, String> peerAddresses = new HashMap<>();
 
         String currentMemberName = this.cluster.getCurrentMemberName();
@@ -695,7 +722,7 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
         for(String memberName : members) {
             if(!currentMemberName.equals(memberName)) {
                 ShardIdentifier shardId = getShardIdentifier(memberName, shardName);
-                String path = getShardActorPath(shardName, currentMemberName);
+                String path = getShardActorPath(shardName, memberName);
                 peerAddresses.put(shardId.toString(), path);
             }
         }
@@ -809,13 +836,25 @@ public class ShardManager extends AbstractUntypedPersistentActorWithMetering {
                                 peerId, peerAddress, actor.path());
                     }
 
-                    actor.tell(new PeerAddressResolved(peerId.toString(), peerAddress), sender);
+                    actor.tell(new PeerAddressResolved(peerId, peerAddress), sender);
                 }
 
                 notifyOnShardInitializedCallbacks();
             }
         }
 
+        void peerDown(String memberName, String peerId, ActorRef sender) {
+            if(peerAddresses.containsKey(peerId) && actor != null) {
+                actor.tell(new PeerDown(memberName, peerId), sender);
+            }
+        }
+
+        void peerUp(String memberName, String peerId, ActorRef sender) {
+            if(peerAddresses.containsKey(peerId) && actor != null) {
+                actor.tell(new PeerUp(memberName, peerId), sender);
+            }
+        }
+
         boolean isShardReady() {
             return !RaftState.Candidate.name().equals(role) && !Strings.isNullOrEmpty(role);
         }