[Goldeneye] CORD-568 Ensure location transparency of node init operation

- Added MapListener for cordvtn node store and made the init process to be
  triggered by map event, so that the leader can do its job regardless of
  the location where node init CLI command happens
- Fixed equals and hashCode override to use all node attributes except for
  the node init state
- Adjusted some log levels

Change-Id: I45688afa60de3516d91132e8a6c49cf90c4dcae4

@@ -718,7 +718,7 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
                 .stream()
                 .stream()
                 .forEach(entry -> {
                 .forEach(entry -> {
                     arpProxy.addGateway(entry.getKey(), entry.getValue());
                     arpProxy.addGateway(entry.getKey(), entry.getValue());
-                    log.debug("Added public gateway IP {}, MAC {}",
+                    log.info("Added public gateway IP {}, MAC {}",
                               entry.getKey().toString(), entry.getValue().toString());
                               entry.getKey().toString(), entry.getValue().toString());
                 });
                 });
         // TODO notice gateway MAC change to VMs holds this gateway IP
         // TODO notice gateway MAC change to VMs holds this gateway IP

@@ -97,7 +97,6 @@ public class CordVtnConfig extends Config<ApplicationId> {
                         getConfig(jsonNode, DATA_PLANE_INTF),
                         getConfig(jsonNode, DATA_PLANE_INTF),
                         CordVtnNodeState.noState());
                         CordVtnNodeState.noState());
 
 
-                log.info("Successfully read {} from the config", hostname);
                 nodes.add(newNode);
                 nodes.add(newNode);
             } catch (IllegalArgumentException | NullPointerException e) {
             } catch (IllegalArgumentException | NullPointerException e) {
                 log.error("{}", e.toString());
                 log.error("{}", e.toString());
@@ -121,7 +120,6 @@ public class CordVtnConfig extends Config<ApplicationId> {
             log.error("{} is not configured", path);
             log.error("{} is not configured", path);
             return null;
             return null;
         } else {
         } else {
-            log.debug("{} : {}", path, jsonNode.asText());
             return jsonNode.asText();
             return jsonNode.asText();
         }
         }
     }
     }

@@ -181,11 +181,16 @@ public final class CordVtnNode {
             return true;
             return true;
         }
         }
 
 
-        // hostname here is a network hostname and it is intended to be
-        // unique throughout the service.
         if (obj instanceof CordVtnNode) {
         if (obj instanceof CordVtnNode) {
             CordVtnNode that = (CordVtnNode) obj;
             CordVtnNode that = (CordVtnNode) obj;
-            if (Objects.equals(hostname, that.hostname)) {
+            if (Objects.equals(hostname, that.hostname) &&
+                    Objects.equals(hostMgmtIp, that.hostMgmtIp) &&
+                    Objects.equals(localMgmtIp, that.localMgmtIp) &&
+                    Objects.equals(dpIp, that.dpIp) &&
+                    Objects.equals(ovsdbPort, that.ovsdbPort) &&
+                    Objects.equals(sshInfo, that.sshInfo) &&
+                    Objects.equals(bridgeId, that.bridgeId) &&
+                    Objects.equals(dpIntf, that.dpIntf)) {
                 return true;
                 return true;
             }
             }
         }
         }
@@ -194,7 +199,8 @@ public final class CordVtnNode {
 
 
     @Override
     @Override
     public int hashCode() {
     public int hashCode() {
-        return Objects.hash(hostname);
+        return Objects.hash(hostname, hostMgmtIp, localMgmtIp, dpIp,
+                            ovsdbPort, sshInfo, bridgeId, dpIntf);
     }
     }
 
 
     @Override
     @Override

@@ -62,6 +62,8 @@ import org.onosproject.ovsdb.controller.OvsdbController;
 import org.onosproject.ovsdb.controller.OvsdbNodeId;
 import org.onosproject.ovsdb.controller.OvsdbNodeId;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.service.ConsistentMap;
 import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.StorageService;
 import org.onosproject.store.service.StorageService;
 import org.onosproject.store.service.Versioned;
 import org.onosproject.store.service.Versioned;
@@ -164,6 +166,7 @@ public class CordVtnNodeManager {
 
 
     private final NetworkConfigListener configListener = new InternalConfigListener();
     private final NetworkConfigListener configListener = new InternalConfigListener();
     private final DeviceListener deviceListener = new InternalDeviceListener();
     private final DeviceListener deviceListener = new InternalDeviceListener();
+    private final MapEventListener<String, CordVtnNode> nodeStoreListener = new InternalMapListener();
 
 
     private final OvsdbHandler ovsdbHandler = new OvsdbHandler();
     private final OvsdbHandler ovsdbHandler = new OvsdbHandler();
     private final BridgeHandler bridgeHandler = new BridgeHandler();
     private final BridgeHandler bridgeHandler = new BridgeHandler();
@@ -236,6 +239,7 @@ public class CordVtnNodeManager {
                                                  configRegistry,
                                                  configRegistry,
                                                  DEFAULT_TUNNEL);
                                                  DEFAULT_TUNNEL);
 
 
+        nodeStore.addListener(nodeStoreListener);
         deviceService.addListener(deviceListener);
         deviceService.addListener(deviceListener);
         configService.addListener(configListener);
         configService.addListener(configListener);
     }
     }
@@ -245,22 +249,21 @@ public class CordVtnNodeManager {
         configService.removeListener(configListener);
         configService.removeListener(configListener);
         deviceService.removeListener(deviceListener);
         deviceService.removeListener(deviceListener);
 
 
-        eventExecutor.shutdown();
+        nodeStore.removeListener(nodeStoreListener);
         nodeStore.clear();
         nodeStore.clear();
+
         leadershipService.withdraw(appId.name());
         leadershipService.withdraw(appId.name());
+        eventExecutor.shutdown();
     }
     }
 
 
     /**
     /**
-     * Adds a new node to the service.
+     * Adds or updates a new node to the service.
      *
      *
      * @param node cordvtn node
      * @param node cordvtn node
      */
      */
-    public void addNode(CordVtnNode node) {
+    public void addOrUpdateNode(CordVtnNode node) {
         checkNotNull(node);
         checkNotNull(node);
-
-        // allow update node attributes
         nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, getNodeState(node)));
         nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, getNodeState(node)));
-        initNode(node);
     }
     }
 
 
     /**
     /**
@@ -283,23 +286,12 @@ public class CordVtnNodeManager {
      *
      *
      * @param node cordvtn node
      * @param node cordvtn node
      */
      */
-    public void initNode(CordVtnNode node) {
+    private void initNode(CordVtnNode node) {
         checkNotNull(node);
         checkNotNull(node);
 
 
-        if (!nodeStore.containsKey(node.hostname())) {
-            log.warn("Node {} does not exist, add node first", node.hostname());
-            return;
-        }
-
-        NodeId leaderNodeId = leadershipService.getLeader(appId.name());
-        log.debug("Node init requested, local: {} leader: {}", localNodeId, leaderNodeId);
-        if (!Objects.equals(localNodeId, leaderNodeId)) {
-            // only the leader performs node init
-            return;
-        }
+        NodeState state = (NodeState) node.state();
+        log.debug("Processing node: {} state: {}", node.hostname(), state);
 
 
-        NodeState state = getNodeState(node);
-        log.debug("Init node: {} state: {}", node.hostname(), state.toString());
         state.process(this, node);
         state.process(this, node);
     }
     }
 
 
@@ -432,9 +424,8 @@ public class CordVtnNodeManager {
     private void setNodeState(CordVtnNode node, NodeState newState) {
     private void setNodeState(CordVtnNode node, NodeState newState) {
         checkNotNull(node);
         checkNotNull(node);
 
 
-        log.debug("Changed {} state: {}", node.hostname(), newState.toString());
+        log.debug("Changed {} state: {}", node.hostname(), newState);
         nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, newState));
         nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, newState));
-        newState.process(this, node);
     }
     }
 
 
     /**
     /**
@@ -812,7 +803,7 @@ public class CordVtnNodeManager {
                 return;
                 return;
             }
             }
 
 
-            log.debug("Port {} is added to {}", portName, node.hostname());
+            log.info("Port {} is added to {}", portName, node.hostname());
 
 
             if (portName.startsWith(VPORT_PREFIX)) {
             if (portName.startsWith(VPORT_PREFIX)) {
                 if (isNodeStateComplete(node)) {
                 if (isNodeStateComplete(node)) {
@@ -840,7 +831,7 @@ public class CordVtnNodeManager {
                 return;
                 return;
             }
             }
 
 
-            log.debug("Port {} is removed from {}", portName, node.hostname());
+            log.info("Port {} is removed from {}", portName, node.hostname());
 
 
             if (portName.startsWith(VPORT_PREFIX)) {
             if (portName.startsWith(VPORT_PREFIX)) {
                 if (isNodeStateComplete(node)) {
                 if (isNodeStateComplete(node)) {
@@ -861,7 +852,7 @@ public class CordVtnNodeManager {
 
 
             NodeId leaderNodeId = leadershipService.getLeader(appId.name());
             NodeId leaderNodeId = leadershipService.getLeader(appId.name());
             if (!Objects.equals(localNodeId, leaderNodeId)) {
             if (!Objects.equals(localNodeId, leaderNodeId)) {
-                // only the leader processes events
+                // do not allow to proceed without leadership
                 return;
                 return;
             }
             }
 
 
@@ -871,19 +862,19 @@ public class CordVtnNodeManager {
 
 
             switch (event.type()) {
             switch (event.type()) {
                 case PORT_ADDED:
                 case PORT_ADDED:
-                    eventExecutor.submit(() -> bridgeHandler.portAdded(event.port()));
+                    eventExecutor.execute(() -> bridgeHandler.portAdded(event.port()));
                     break;
                     break;
                 case PORT_UPDATED:
                 case PORT_UPDATED:
                     if (!event.port().isEnabled()) {
                     if (!event.port().isEnabled()) {
-                        eventExecutor.submit(() -> bridgeHandler.portRemoved(event.port()));
+                        eventExecutor.execute(() -> bridgeHandler.portRemoved(event.port()));
                     }
                     }
                     break;
                     break;
                 case DEVICE_ADDED:
                 case DEVICE_ADDED:
                 case DEVICE_AVAILABILITY_CHANGED:
                 case DEVICE_AVAILABILITY_CHANGED:
                     if (deviceService.isAvailable(device.id())) {
                     if (deviceService.isAvailable(device.id())) {
-                        eventExecutor.submit(() -> handler.connected(device));
+                        eventExecutor.execute(() -> handler.connected(device));
                     } else {
                     } else {
-                        eventExecutor.submit(() -> handler.disconnected(device));
+                        eventExecutor.execute(() -> handler.disconnected(device));
                     }
                     }
                     break;
                     break;
                 default:
                 default:
@@ -902,14 +893,19 @@ public class CordVtnNodeManager {
             return;
             return;
         }
         }
 
 
-        config.cordVtnNodes().forEach(this::addNode);
-        // TODO remove nodes if needed
+        config.cordVtnNodes().forEach(this::addOrUpdateNode);
     }
     }
 
 
     private class InternalConfigListener implements NetworkConfigListener {
     private class InternalConfigListener implements NetworkConfigListener {
 
 
         @Override
         @Override
         public void event(NetworkConfigEvent event) {
         public void event(NetworkConfigEvent event) {
+            NodeId leaderNodeId = leadershipService.getLeader(appId.name());
+            if (!Objects.equals(localNodeId, leaderNodeId)) {
+                // do not allow to proceed without leadership
+                return;
+            }
+
             if (!event.configClass().equals(CordVtnConfig.class)) {
             if (!event.configClass().equals(CordVtnConfig.class)) {
                 return;
                 return;
             }
             }
@@ -924,4 +920,46 @@ public class CordVtnNodeManager {
             }
             }
         }
         }
     }
     }
+
+    private class InternalMapListener implements MapEventListener<String, CordVtnNode> {
+
+        @Override
+        public void event(MapEvent<String, CordVtnNode> event) {
+            NodeId leaderNodeId = leadershipService.getLeader(appId.name());
+            if (!Objects.equals(localNodeId, leaderNodeId)) {
+                // do not allow to proceed without leadership
+                return;
+            }
+
+            CordVtnNode oldNode;
+            CordVtnNode newNode;
+
+            switch (event.type()) {
+                case UPDATE:
+                    oldNode = event.oldValue().value();
+                    newNode = event.newValue().value();
+
+                    if (!newNode.equals(oldNode)) {
+                        log.info("{} has been updated", newNode.hostname());
+                        log.debug("New node: {}", newNode);
+                    }
+                    // perform init procedure based on current state on any updates,
+                    // insert, or even if the node is the same for robustness since
+                    // it's no harm to run the init procedure multiple times
+                    eventExecutor.execute(() -> initNode(newNode));
+                    break;
+                case INSERT:
+                    newNode = event.newValue().value();
+                    log.info("Added {}", newNode.hostname());
+                    eventExecutor.execute(() -> initNode(newNode));
+                    break;
+                case REMOVE:
+                    oldNode = event.oldValue().value();
+                    log.info("{} is removed", oldNode.hostname());
+                    break;
+                default:
+                    break;
+            }
+        }
+    }
 }
 }

@@ -51,7 +51,7 @@ public class CordVtnNodeInitCommand extends AbstractShellCommand {
                 continue;
                 continue;
             }
             }
 
 
-            nodeManager.initNode(node);
+            nodeManager.addOrUpdateNode(node);
         }
         }
     }
     }
 }
 }