OpenFlow Protocol_plugin changes to make use of the Connection Manager infrastructure...

Active-Active Clustering support. Notable changes are :
1. Interested Protocol_plugin services registers with SAL for Connection Service.
2. isLocal API is used by InventoryServiceShim to decide whether or not to advertise the learnt node on a particular controller.
3. isLocal API is used by other services to handle node specific events
4. InventoryService instance can now represent Global inventory in addition to its additional instances being container specific
5. Controller provides the disconnect API to allow the Connection manager disconnect a switch. (Connect API is stubbed out)

Change-Id: I163fa7b14fa998bb4654b01c0a904c1fec09b0bb
Signed-off-by: Madhu Venugopal <vmadhu@cisco.com>

diff --git a/opendaylight/protocol_plugins/openflow/pom.xml b/opendaylight/protocol_plugins/openflow/pom.xml
index 9d84d45967a44114a960dfc4ab3cd56f671e2dfd..373d67dab15976a117d15a989d716948cf42621b 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/pom.xml
+++ b/opendaylight/protocol_plugins/openflow/pom.xml
@@ -32,6 +32,7 @@
               org.opendaylight.controller.sal.inventory,
               org.opendaylight.controller.sal.match,
               org.opendaylight.controller.sal.utils,
+              org.opendaylight.controller.sal.connection,
               org.apache.commons.lang3.builder,
               org.apache.commons.lang3.tuple,
               org.apache.felix.dm,
@@ -64,6 +65,11 @@
       <artifactId>sal</artifactId>
       <version>0.5.0-SNAPSHOT</version>
     </dependency>
+    <dependency>
+      <groupId>org.opendaylight.controller</groupId>
+      <artifactId>sal.connection</artifactId>
+      <version>0.1.0-SNAPSHOT</version>
+    </dependency>
     <dependency>
       <groupId>org.opendaylight.controller.thirdparty</groupId>
       <artifactId>org.openflow.openflowj</artifactId>

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/core/internal/Controller.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/core/internal/Controller.java
index 40f594bd2a0e12bfe13215fb50672909fe5de856..c7c6c8924d29807da61aa5896c4b2b1e1b79cc59 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/core/internal/Controller.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/core/internal/Controller.java
@@ -30,6 +30,12 @@ import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitchStateListener;
+import org.opendaylight.controller.sal.connection.ConnectionConstants;
+import org.opendaylight.controller.sal.connection.IPluginInConnectionService;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
+import org.opendaylight.controller.sal.core.Node;
+import org.opendaylight.controller.sal.utils.Status;
+import org.opendaylight.controller.sal.utils.StatusCode;
 import org.openflow.protocol.OFMessage;
 import org.openflow.protocol.OFType;
 import org.openflow.util.HexString;
@@ -38,7 +44,7 @@ import org.osgi.framework.FrameworkUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class Controller implements IController, CommandProvider {
+public class Controller implements IController, CommandProvider, IPluginInConnectionService {
     private static final Logger logger = LoggerFactory
             .getLogger(Controller.class);
     private ControllerIO controllerIO;
@@ -221,8 +227,7 @@ public class Controller implements IController, CommandProvider {
             // create new switch
             int i = this.switchInstanceNumber.addAndGet(1);
             String instanceName = "SwitchHandler-" + i;
-            SwitchHandler switchHandler = new SwitchHandler(this, sc,
-                    instanceName);
+            SwitchHandler switchHandler = new SwitchHandler(this, sc, instanceName);
             switchHandler.start();
             if (sc.isConnected()) {
                 logger.info("Switch:{} is connected to the Controller",
@@ -375,4 +380,34 @@ public class Controller implements IController, CommandProvider {
         help.append("\t controllerShowConnConfig\n");
         return help.toString();
     }
+
+    @Override
+    public Status disconnect(Node node) {
+        ISwitch sw = getSwitch((Long) node.getID());
+        if (sw != null) disconnectSwitch(sw);
+        return new Status(StatusCode.SUCCESS);
+    }
+
+    @Override
+    public Node connect(String connectionIdentifier, Map<ConnectionConstants, String> params) {
+        return null;
+    }
+
+    /**
+     * View Change notification
+     */
+    public void notifyClusterViewChanged() {
+        for (ISwitch sw : switches.values()) {
+            notifySwitchAdded(sw);
+        }
+    }
+
+    /**
+     * Node Disconnected from the node's master controller.
+     */
+    @Override
+    public void notifyNodeDisconnectFromMaster(Node node) {
+        ISwitch sw = switches.get((Long)node.getID());
+        if (sw != null) notifySwitchAdded(sw);
+    }
 }

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/Activator.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/Activator.java
index 6c09abbdc7cd851df41e9e72b2bba987cd6171e9..16393fa21f0518fed7691f0960fbd53c02b54ce0 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/Activator.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/Activator.java
@@ -28,6 +28,8 @@ import org.opendaylight.controller.protocol_plugin.openflow.ITopologyServiceShim
 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
 import org.opendaylight.controller.protocol_plugin.openflow.core.internal.Controller;
+import org.opendaylight.controller.sal.connection.IPluginInConnectionService;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.ComponentActivatorAbstractBase;
 import org.opendaylight.controller.sal.core.IContainerListener;
 import org.opendaylight.controller.sal.core.Node;
@@ -162,6 +164,11 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setCallbacks("setPluginOutDataPacketService",
                             "unsetPluginOutDataPacketService")
                     .setRequired(false));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
 
         if (imp.equals(ReadService.class)) {
@@ -178,11 +185,18 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setService(IReadServiceFilter.class)
                     .setCallbacks("setService", "unsetService")
                     .setRequired(true));
+
             c.add(createContainerServiceDependency(containerName)
                     .setService(IPluginOutReadService.class)
                     .setCallbacks("setPluginOutReadServices",
                             "unsetPluginOutReadServices")
                     .setRequired(false));
+
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
 
         if (imp.equals(FlowProgrammerNotifier.class)) {
@@ -198,6 +212,11 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setCallbacks("setPluginOutFlowProgrammerService",
                             "unsetPluginOutFlowProgrammerService")
                     .setRequired(true));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
     }
 
@@ -213,7 +232,7 @@ public class Activator extends ComponentActivatorAbstractBase {
     public Object[] getGlobalImplementations() {
         Object[] res = { Controller.class, OFStatisticsManager.class,
                 FlowProgrammerService.class, ReadServiceFilter.class,
-                DiscoveryService.class, DataPacketMuxDemux.class,
+                DiscoveryService.class, DataPacketMuxDemux.class, InventoryService.class,
                 InventoryServiceShim.class, TopologyServiceShim.class };
         return res;
     }
@@ -235,7 +254,10 @@ public class Activator extends ComponentActivatorAbstractBase {
             logger.debug("Activator configureGlobalInstance( ) is called");
             Dictionary<String, Object> props = new Hashtable<String, Object>();
             props.put("name", "Controller");
-            c.setInterface(IController.class.getName(), props);
+            props.put(GlobalConstants.PROTOCOLPLUGINTYPE.toString(), Node.NodeIDType.OPENFLOW);
+            c.setInterface(new String[] { IController.class.getName(),
+                                          IPluginInConnectionService.class.getName()},
+                                          props);
         }
 
         if (imp.equals(FlowProgrammerService.class)) {
@@ -263,6 +285,11 @@ public class Activator extends ComponentActivatorAbstractBase {
                             "unsetsetFlowProgrammerNotifier")
                     .setRequired(false));
 
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
 
         if (imp.equals(ReadServiceFilter.class)) {
@@ -285,7 +312,6 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setCallbacks("setReadFilterInternalListener",
                             "unsetReadFilterInternalListener")
                     .setRequired(false));
-
         }
 
         if (imp.equals(OFStatisticsManager.class)) {
@@ -327,6 +353,11 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setService(IDiscoveryListener.class)
                     .setCallbacks("setDiscoveryListener",
                             "unsetDiscoveryListener").setRequired(true));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
 
         // DataPacket mux/demux services, which is teh actual engine
@@ -350,6 +381,34 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setService(IDataPacketListen.class)
                     .setCallbacks("setIDataPacketListen",
                             "unsetIDataPacketListen").setRequired(false));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
+        }
+
+        if (imp.equals(InventoryService.class)) {
+            // export the service
+            Dictionary<String, Object> props = new Hashtable<String, Object>();
+            props.put("scope", "Global");
+
+            c.setInterface(
+                    new String[] { IPluginInInventoryService.class.getName(),
+                            IInventoryShimInternalListener.class.getName(),
+                            IInventoryProvider.class.getName() }, props);
+
+            // Now lets add a service dependency to make sure the
+            // provider of service exists
+            c.add(createServiceDependency()
+                    .setService(IController.class, "(name=Controller)")
+                    .setCallbacks("setController", "unsetController")
+                    .setRequired(true));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutInventoryService.class, "(scope=Global)")
+                    .setCallbacks("setPluginOutInventoryServices",
+                            "unsetPluginOutInventoryServices")
+                    .setRequired(false));
         }
 
         if (imp.equals(InventoryServiceShim.class)) {
@@ -361,15 +420,25 @@ public class Activator extends ComponentActivatorAbstractBase {
                     .setCallbacks("setController", "unsetController")
                     .setRequired(true));
             c.add(createServiceDependency()
-                    .setService(IInventoryShimInternalListener.class)
+                    .setService(IInventoryShimInternalListener.class, "(!(scope=Global))")
                     .setCallbacks("setInventoryShimInternalListener",
                             "unsetInventoryShimInternalListener")
                     .setRequired(true));
+            c.add(createServiceDependency()
+                    .setService(IInventoryShimInternalListener.class, "(scope=Global)")
+                    .setCallbacks("setInventoryShimGlobalInternalListener",
+                            "unsetInventoryShimGlobalInternalListener")
+                    .setRequired(true));
             c.add(createServiceDependency()
                     .setService(IInventoryShimExternalListener.class)
                     .setCallbacks("setInventoryShimExternalListener",
                             "unsetInventoryShimExternalListener")
                     .setRequired(false));
+            c.add(createServiceDependency()
+                    .setService(IPluginOutConnectionService.class)
+                    .setCallbacks("setIPluginOutConnectionService",
+                            "unsetIPluginOutConnectionService")
+                    .setRequired(false));
         }
 
         if (imp.equals(TopologyServiceShim.class)) {

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
index 2d8cdd206a7e2c651151c6515e003ba8e862c8fa..7a01d3bd19bce2060a6749bb017e08026a84aaea 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketMuxDemux.java
@@ -32,6 +32,7 @@ import org.openflow.protocol.action.OFActionOutput;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.ConstructionException;
 import org.opendaylight.controller.sal.core.ContainerFlow;
 import org.opendaylight.controller.sal.core.IContainerListener;
@@ -60,6 +61,7 @@ public class DataPacketMuxDemux implements IContainerListener,
     private ConcurrentMap<String, List<ContainerFlow>> container2FlowSpecs = new ConcurrentHashMap<String, List<ContainerFlow>>();
     // Track local data packet listener
     private List<IDataPacketListen> iDataPacketListen = new CopyOnWriteArrayList<IDataPacketListen>();
+    private IPluginOutConnectionService connectionOutService;
 
     void setIDataPacketListen(IDataPacketListen s) {
         if (this.iDataPacketListen != null) {
@@ -128,6 +130,16 @@ public class DataPacketMuxDemux implements IContainerListener,
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     /**
      * Function called by the dependency manager when all the required
      * dependencies are satisfied
@@ -164,9 +176,21 @@ public class DataPacketMuxDemux implements IContainerListener,
                     new Object[] { sw, msg, this.pluginOutDataPacketServices });
             return;
         }
+
+        Long ofSwitchID = Long.valueOf(sw.getId());
+        try {
+            Node n = new Node(Node.NodeIDType.OPENFLOW, ofSwitchID);
+            if (!connectionOutService.isLocal(n)) {
+                logger.debug("Connection service refused DataPacketMuxDemux receive {} {}", sw, msg);
+                return;
+            }
+        }
+        catch (Exception e) {
+            return;
+        }
+
         if (msg instanceof OFPacketIn) {
             OFPacketIn ofPacket = (OFPacketIn) msg;
-            Long ofSwitchID = Long.valueOf(sw.getId());
             Short ofPortID = Short.valueOf(ofPacket.getInPort());
 
             try {
@@ -258,6 +282,12 @@ public class DataPacketMuxDemux implements IContainerListener,
             return;
         }
 
+        if (!connectionOutService.isLocal(outPort.getNode())) {
+            logger.debug("data packets will not be sent to {} in a non-master controller", outPort.toString());
+            return;
+        }
+
+
         if (!outPort.getType().equals(
                 NodeConnector.NodeConnectorIDType.OPENFLOW)) {
             // The output Port is not of type OpenFlow

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketServices.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketServices.java
index adb9d20ef7d4338a287a6e721bbb23e9f034fca5..4c65645bfed8f98c53bc019e8351e7c2b09cec01 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketServices.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DataPacketServices.java
@@ -12,6 +12,8 @@ package org.opendaylight.controller.protocol_plugin.openflow.internal;
 import org.opendaylight.controller.protocol_plugin.openflow.IDataPacketMux;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
+import org.opendaylight.controller.sal.core.NodeConnector;
 import org.opendaylight.controller.sal.packet.IPluginInDataPacketService;
 import org.opendaylight.controller.sal.packet.RawPacket;
 
@@ -19,6 +21,7 @@ public class DataPacketServices implements IPluginInDataPacketService {
     protected static final Logger logger = LoggerFactory
             .getLogger(DataPacketServices.class);
     private IDataPacketMux iDataPacketMux = null;
+    private IPluginOutConnectionService connectionOutService;
 
     void setIDataPacketMux(IDataPacketMux s) {
         this.iDataPacketMux = s;
@@ -30,8 +33,23 @@ public class DataPacketServices implements IPluginInDataPacketService {
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     @Override
     public void transmitDataPacket(RawPacket outPkt) {
-        this.iDataPacketMux.transmitDataPacket(outPkt);
+        NodeConnector nc = outPkt.getOutgoingNodeConnector();
+        if (connectionOutService != null && connectionOutService.isLocal(nc.getNode())) {
+            this.iDataPacketMux.transmitDataPacket(outPkt);
+        } else {
+            logger.debug("{} is dropped in the controller "+outPkt);
+        }
     }
 }

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DiscoveryService.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DiscoveryService.java
index ee712030e95573a807239dc6283f4291bb5f01d0..bb303e3651de7e3a95659641cf001da4b0153bb4 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DiscoveryService.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/DiscoveryService.java
@@ -39,6 +39,7 @@ import org.osgi.framework.FrameworkUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Config;
 import org.opendaylight.controller.sal.core.ConstructionException;
 import org.opendaylight.controller.sal.core.Edge;
@@ -60,6 +61,8 @@ import org.opendaylight.controller.sal.utils.HexEncode;
 import org.opendaylight.controller.sal.utils.NetUtils;
 import org.opendaylight.controller.sal.utils.NodeConnectorCreator;
 import org.opendaylight.controller.sal.utils.NodeCreator;
+import org.opendaylight.controller.sal.utils.Status;
+import org.opendaylight.controller.sal.utils.StatusCode;
 
 /**
  * The class describes neighbor discovery service for an OpenFlow network.
@@ -126,6 +129,7 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
     private volatile Boolean shuttingDown = false;
 
     private LLDPTLV chassisIdTlv, portIdTlv, ttlTlv, customTlv;
+    private IPluginOutConnectionService connectionOutService;
 
     class DiscoveryTransmit implements Runnable {
         private final BlockingQueue<NodeConnector> transmitQ;
@@ -256,6 +260,11 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
             return;
         }
 
+        if (!connectionOutService.isLocal(nodeConnector.getNode())) {
+            logger.debug("Discoery packets will not be sent to {} in a non-master controller", nodeConnector.toString());
+            return;
+        }
+
         if (outPkt == null) {
             logger.debug("Can not send discovery packet out since outPkt is null");
             return;
@@ -301,12 +310,18 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
             return PacketResult.IGNORED;
         }
 
-        if (((Short) inPkt.getIncomingNodeConnector().getID()).equals(NodeConnector.SPECIALNODECONNECTORID)) {
+        NodeConnector nodeConnector = inPkt.getIncomingNodeConnector();
+        if (((Short) nodeConnector.getID()).equals(NodeConnector.SPECIALNODECONNECTORID)) {
             logger.trace("Ignoring ethernet packet received on special port: "
                     + inPkt.getIncomingNodeConnector().toString());
             return PacketResult.IGNORED;
         }
 
+        if (!connectionOutService.isLocal(nodeConnector.getNode())) {
+            logger.debug("Discoery packets will not be processed from {} in a non-master controller", nodeConnector.toString());
+            return PacketResult.IGNORED;
+        }
+
         Ethernet ethPkt = new Ethernet();
         try {
             ethPkt.deserialize(data, 0, data.length * NetUtils.NumBitsInAByte);
@@ -659,6 +674,9 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
                 // Allow one more retry
                 readyListLo.add(nodeConnector);
                 elapsedTime.remove(nodeConnector);
+                if (connectionOutService.isLocal(nodeConnector.getNode())) {
+                    transmitQ.add(nodeConnector);
+                }
             }
         }
     }
@@ -692,10 +710,12 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
     private void doDiscovery() {
         if (++discoveryTimerTickCount <= discoveryBatchPauseTicks) {
             for (NodeConnector nodeConnector : getWorkingSet()) {
-                transmitQ.add(nodeConnector);
-                // Move to staging area after it's served
-                if (!stagingList.contains(nodeConnector)) {
-                    stagingList.add(nodeConnector);
+                if (connectionOutService.isLocal(nodeConnector.getNode())) {
+                    transmitQ.add(nodeConnector);
+                    // Move to staging area after it's served
+                    if (!stagingList.contains(nodeConnector)) {
+                        stagingList.add(nodeConnector);
+                    }
                 }
             }
         } else if (discoveryTimerTickCount >= discoveryBatchRestartTicks) {
@@ -1402,6 +1422,16 @@ public class DiscoveryService implements IInventoryShimExternalListener, IDataPa
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     private void initDiscoveryPacket() {
         // Create LLDP ChassisID TLV
         chassisIdTlv = new LLDPTLV();

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerNotifier.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerNotifier.java
index 2a62d6c34a08a19c4cccb898acdf90945ada255b..dfa2026cddc041ea08b040c33b50ca31de0cc8ca 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerNotifier.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerNotifier.java
@@ -10,6 +10,7 @@ package org.opendaylight.controller.protocol_plugin.openflow.internal;
 
 import org.apache.felix.dm.Component;
 import org.opendaylight.controller.protocol_plugin.openflow.IFlowProgrammerNotifier;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Node;
 import org.opendaylight.controller.sal.flowprogrammer.Flow;
 import org.opendaylight.controller.sal.flowprogrammer.IPluginOutFlowProgrammerService;
@@ -24,6 +25,7 @@ public class FlowProgrammerNotifier implements IFlowProgrammerNotifier {
     protected static final Logger logger = LoggerFactory
             .getLogger(FlowProgrammerNotifier.class);
     private IPluginOutFlowProgrammerService salNotifier;
+    private IPluginOutConnectionService connectionOutService;
 
     public FlowProgrammerNotifier() {
         salNotifier = null;
@@ -76,6 +78,11 @@ public class FlowProgrammerNotifier implements IFlowProgrammerNotifier {
 
     @Override
     public void flowRemoved(Node node, Flow flow) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("flow removed will not be notified in a non-master controller for node "+node);
+            return;
+        }
+
         if (salNotifier != null) {
             salNotifier.flowRemoved(node, flow);
         } else {
@@ -85,6 +92,11 @@ public class FlowProgrammerNotifier implements IFlowProgrammerNotifier {
 
     @Override
     public void flowErrorReported(Node node, long rid, Object err) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("flow error will not be notified in a non-master controller for node "+node);
+            return;
+        }
+
         if (salNotifier != null) {
             salNotifier.flowErrorReported(node, rid, err);
         } else {
@@ -92,4 +104,14 @@ public class FlowProgrammerNotifier implements IFlowProgrammerNotifier {
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
 }

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerService.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerService.java
index f58acf62edb0962703b9e4971e582bc412314c47..c55a88cd3f0224cc587bc01e632e742405cc34f1 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerService.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerService.java
@@ -23,6 +23,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExtern
 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.ContainerFlow;
 import org.opendaylight.controller.sal.core.IContainerListener;
 import org.opendaylight.controller.sal.core.Node;
@@ -65,6 +66,7 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
     private Map<String, Set<NodeConnector>> containerToNc;
     private ConcurrentMap<Long, Map<Integer, Long>> xid2rid;
     private int barrierMessagePriorCount = getBarrierMessagePriorCount();
+    private IPluginOutConnectionService connectionOutService;
 
     public FlowProgrammerService() {
         controller = null;
@@ -83,6 +85,16 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     public void setFlowProgrammerNotifier(Map<String, ?> props,
             IFlowProgrammerNotifier s) {
         if (props == null || props.get("containerName") == null) {
@@ -146,32 +158,62 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
 
     @Override
     public Status addFlow(Node node, Flow flow) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Add flow will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return addFlowInternal(node, flow, 0);
     }
 
     @Override
     public Status modifyFlow(Node node, Flow oldFlow, Flow newFlow) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Modify flow will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return modifyFlowInternal(node, oldFlow, newFlow, 0);
     }
 
     @Override
     public Status removeFlow(Node node, Flow flow) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Remove flow will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return removeFlowInternal(node, flow, 0);
     }
 
     @Override
     public Status addFlowAsync(Node node, Flow flow, long rid) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Add flow Async will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return addFlowInternal(node, flow, rid);
     }
 
     @Override
     public Status modifyFlowAsync(Node node, Flow oldFlow, Flow newFlow,
             long rid) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Modify flow async will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return modifyFlowInternal(node, oldFlow, newFlow, rid);
     }
 
     @Override
     public Status removeFlowAsync(Node node, Flow flow, long rid) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Remove flow async will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return removeFlowInternal(node, flow, rid);
     }
 
@@ -324,6 +366,11 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
 
     @Override
     public Status removeAllFlows(Node node) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Remove all flows will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         return new Status(StatusCode.SUCCESS);
     }
 
@@ -446,6 +493,11 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
 
     @Override
     public Status syncSendBarrierMessage(Node node) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("Sync Send Barrier will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
             return new Status(StatusCode.NOTACCEPTABLE,
                     "The node does not support Barrier message.");
@@ -469,6 +521,11 @@ public class FlowProgrammerService implements IPluginInFlowProgrammerService,
 
     @Override
     public Status asyncSendBarrierMessage(Node node) {
+        if (!connectionOutService.isLocal(node)) {
+            log.debug("ASync Send Barrier will not be processed in a non-master controller for node " + node);
+            return new Status(StatusCode.NOTALLOWED, "This is not the master controller for " + node);
+        }
+
         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
             return new Status(StatusCode.NOTACCEPTABLE,
                     "The node does not support Barrier message.");

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryService.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryService.java
index 6e6cb00f899d13a94188f7c76c439a68b331f573..9fded15f42383a1edceb436f5c8b010909a6e03b 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryService.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryService.java
@@ -22,6 +22,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.IInventoryProvider;
 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimInternalListener;
 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Node;
 import org.opendaylight.controller.sal.core.NodeConnector;
 import org.opendaylight.controller.sal.core.Property;
@@ -73,8 +74,10 @@ public class InventoryService implements IInventoryShimInternalListener,
         Dictionary props = c.getServiceProperties();
         if (props != null) {
             containerName = (String) props.get("containerName");
-            isDefaultContainer = containerName.equals(GlobalConstants.DEFAULT
-                    .toString());
+            if (containerName != null) {
+                isDefaultContainer = containerName.equals(GlobalConstants.DEFAULT
+                        .toString());
+            }
         }
 
         nodeProps = new ConcurrentHashMap<Node, Map<String, Property>>();
@@ -201,13 +204,6 @@ public class InventoryService implements IInventoryShimInternalListener,
             return;
         }
 
-        Set<Node> nodeSet = nodeProps.keySet();
-        if (((props == null) || props.isEmpty()) && (nodeSet != null)
-                && nodeSet.contains(node)) {
-            // node already added
-            return;
-        }
-
         logger.trace("addNode: {} added, props: {} for container {}",
                 new Object[] { node, props, containerName });
 
@@ -297,5 +293,4 @@ public class InventoryService implements IInventoryShimInternalListener,
             break;
         }
     }
-
 }

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryServiceShim.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryServiceShim.java
index f0b8735f83f960b0626fff87c20304ce40a03cb9..241fa92deb267414704bb8f2dca5095cf8078d05 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryServiceShim.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/InventoryServiceShim.java
@@ -26,6 +26,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitchStateListener;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Actions;
 import org.opendaylight.controller.sal.core.Buffers;
 import org.opendaylight.controller.sal.core.Capabilities;
@@ -63,11 +64,13 @@ public class InventoryServiceShim implements IContainerListener,
             .getLogger(InventoryServiceShim.class);
     private IController controller = null;
     private final ConcurrentMap<String, IInventoryShimInternalListener> inventoryShimInternalListeners = new ConcurrentHashMap<String, IInventoryShimInternalListener>();
+    private final Set<IInventoryShimInternalListener> globalInventoryShimInternalListeners = new HashSet<IInventoryShimInternalListener>();
     private final List<IInventoryShimExternalListener> inventoryShimExternalListeners = new CopyOnWriteArrayList<IInventoryShimExternalListener>();
     private final ConcurrentMap<NodeConnector, Set<String>> nodeConnectorContainerMap = new ConcurrentHashMap<NodeConnector, Set<String>>();
     private final ConcurrentMap<Node, Set<String>> nodeContainerMap = new ConcurrentHashMap<Node, Set<String>>();
     private final ConcurrentMap<NodeConnector, Set<Property>> nodeConnectorProps = new ConcurrentHashMap<NodeConnector, Set<Property>>();
     private final ConcurrentMap<Node, Set<Property>> nodeProps = new ConcurrentHashMap<Node, Set<Property>>();
+    private IPluginOutConnectionService connectionOutService;
 
     void setController(IController s) {
         this.controller = s;
@@ -79,6 +82,20 @@ public class InventoryServiceShim implements IContainerListener,
         }
     }
 
+    void setInventoryShimGlobalInternalListener(Map<?, ?> props,
+            IInventoryShimInternalListener s) {
+        if ((this.globalInventoryShimInternalListeners != null)) {
+            this.globalInventoryShimInternalListeners.add(s);
+        }
+    }
+
+    void unsetInventoryShimGlobalInternalListener(Map<?, ?> props,
+            IInventoryShimInternalListener s) {
+        if ((this.globalInventoryShimInternalListeners != null)) {
+            this.globalInventoryShimInternalListeners.remove(s);
+        }
+    }
+
     void setInventoryShimInternalListener(Map<?, ?> props,
             IInventoryShimInternalListener s) {
         if (props == null) {
@@ -107,7 +124,7 @@ public class InventoryServiceShim implements IContainerListener,
         }
         String containerName = (String) props.get("containerName");
         if (containerName == null) {
-            logger.error("unsetInventoryShimInternalListener containerName not supplied");
+            logger.error("setInventoryShimInternalListener containerName not supplied");
             return;
         }
         if ((this.inventoryShimInternalListeners != null)
@@ -136,6 +153,16 @@ public class InventoryServiceShim implements IContainerListener,
         }
     }
 
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
+
     /**
      * Function called by the dependency manager when all the required
      * dependencies are satisfied
@@ -167,6 +194,7 @@ public class InventoryServiceShim implements IContainerListener,
         this.inventoryShimInternalListeners.clear();
         this.nodeConnectorContainerMap.clear();
         this.nodeContainerMap.clear();
+        this.globalInventoryShimInternalListeners.clear();
         this.controller = null;
     }
 
@@ -353,6 +381,19 @@ public class InventoryServiceShim implements IContainerListener,
      * Notify all internal and external listeners
      */
     private void notifyInventoryShimListener(NodeConnector nodeConnector, UpdateType type, Set<Property> props) {
+        notifyGlobalInventoryShimInternalListener(nodeConnector, type, props);
+        /*
+         * isLocal is intentionally moved after the GlobalInventory listener call.
+         * The above notification to GlobalInventory will make sure that the connectionOutService be ready
+         * to reply to isLocal query.
+         */
+        if (!connectionOutService.isLocal(nodeConnector.getNode())) {
+            logger.debug("Connection service dropped the inventory notification for {} {}", nodeConnector.toString(), type);
+            return;
+        } else {
+            logger.debug("Connection service accepted the inventory notification for {} {}", nodeConnector.toString(), type);
+        }
+
         // notify other containers
         Set<String> containers = (nodeConnectorContainerMap.get(nodeConnector) == null) ? new HashSet<String>()
                 : new HashSet<String>(nodeConnectorContainerMap.get(nodeConnector));
@@ -369,7 +410,19 @@ public class InventoryServiceShim implements IContainerListener,
      * Notify all internal and external listeners
      */
     private void notifyInventoryShimListener(Node node, UpdateType type, Set<Property> props) {
-        // Now notify other containers
+        notifyGlobalInventoryShimInternalListener(node, type, props);
+        /*
+         * isLocal is intentionally moved after the GlobalInventory listener call.
+         * The above notification to GlobalInventory will make sure that the connectionOutService be ready
+         * to reply to isLocal query.
+         */
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("Connection service dropped the inventory notification for {} {}", node.toString(), type);
+            return;
+        } else {
+            logger.debug("Connection service accepted the inventory notification for {} {}", node.toString(), type);
+        }
+    // Now notify other containers
         Set<String> containers = (nodeContainerMap.get(node) == null) ? new HashSet<String>() : new HashSet<String>(
                 nodeContainerMap.get(node));
         containers.add(GlobalConstants.DEFAULT.toString());
@@ -381,6 +434,24 @@ public class InventoryServiceShim implements IContainerListener,
         notifyInventoryShimExternalListener(node, type, props);
     }
 
+    private void notifyGlobalInventoryShimInternalListener(Node node, UpdateType type, Set<Property> props) {
+        for (IInventoryShimInternalListener globalListener : globalInventoryShimInternalListeners) {
+            globalListener.updateNode(node, type, props);
+            logger.trace(
+                    "notifyGlobalInventoryShimInternalListener {} type {}",
+                    new Object[] { node, type });
+        }
+    }
+
+    private void notifyGlobalInventoryShimInternalListener(NodeConnector nodeConnector, UpdateType type, Set<Property> props) {
+        for (IInventoryShimInternalListener globalListener : globalInventoryShimInternalListeners) {
+            globalListener.updateNodeConnector(nodeConnector, type, props);
+            logger.trace(
+                    "notifyGlobalInventoryShimInternalListener {} type {}",
+                    new Object[] { nodeConnector, type });
+        }
+    }
+
     private void notifyInventoryShimInternalListener(String container,
             Node node, UpdateType type, Set<Property> props) {
         IInventoryShimInternalListener inventoryShimInternalListener = inventoryShimInternalListeners

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/OFStatisticsManager.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/OFStatisticsManager.java
index bae49d511a2f853f61d428f0751ea7b231539831..e3a1ffec96cc28d26b0dcc247763aeef1c06184d 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/OFStatisticsManager.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/OFStatisticsManager.java
@@ -36,6 +36,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6Match;
 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsReply;
 import org.opendaylight.controller.protocol_plugin.openflow.vendorextension.v6extension.V6StatsRequest;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Node;
 import org.opendaylight.controller.sal.core.NodeConnector;
 import org.opendaylight.controller.sal.core.Property;
@@ -169,6 +170,18 @@ IInventoryShimExternalListener, CommandProvider {
         }
         return statsQueueSize;
     }
+
+    IPluginOutConnectionService connectionPluginOutService;
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionPluginOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionPluginOutService == s) {
+            connectionPluginOutService = null;
+        }
+    }
+
     /**
      * Function called by the dependency manager when all the required
      * dependencies are satisfied

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadService.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadService.java
index 585e4f32c46d83b1edbe758ad2b635655639dcc8..84f7dd9282ca89cc67eb160a6465712c2b489d8b 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadService.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadService.java
@@ -17,6 +17,7 @@ import java.util.concurrent.CopyOnWriteArraySet;
 import org.apache.felix.dm.Component;
 import org.opendaylight.controller.protocol_plugin.openflow.IReadFilterInternalListener;
 import org.opendaylight.controller.protocol_plugin.openflow.IReadServiceFilter;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Node;
 import org.opendaylight.controller.sal.core.Node.NodeIDType;
 import org.opendaylight.controller.sal.core.NodeConnector;
@@ -40,6 +41,7 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
     private IReadServiceFilter filter;
     private Set<IPluginOutReadService> pluginOutReadServices;
     private String containerName;
+    private IPluginOutConnectionService connectionOutService;
 
     /**
      * Function called by the dependency manager when all the required
@@ -110,6 +112,10 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             return null;
         }
 
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for the node : " + node);
+            return null;
+        }
         return filter.readFlow(containerName, node, flow, cached);
     }
 
@@ -120,6 +126,11 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             return null;
         }
 
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for the node : " + node);
+            return null;
+        }
+
         return filter.readAllFlow(containerName, node, cached);
     }
 
@@ -130,6 +141,11 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             return null;
         }
 
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for the node : " + node);
+            return null;
+        }
+
         return filter.readDescription(node, cached);
     }
 
@@ -141,6 +157,12 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             logger.error("Invalid node type");
             return null;
         }
+
+        if (!connectionOutService.isLocal(connector.getNode())) {
+            logger.debug("This Controller is not the master for connector : "+connector);
+            return null;
+        }
+
         return filter.readNodeConnector(containerName, connector, cached);
     }
 
@@ -152,6 +174,11 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             return null;
         }
 
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return null;
+        }
+
         return filter.readAllNodeConnector(containerName, node, cached);
     }
 
@@ -162,6 +189,12 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             logger.error("Invalid node type");
             return 0;
         }
+
+        if (!connectionOutService.isLocal(connector.getNode())) {
+            logger.debug("This Controller is not the master for connector : "+connector);
+            return 0;
+        }
+
         return filter.getTransmitRate(containerName, connector);
     }
 
@@ -172,6 +205,12 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             logger.error("Invalid node type");
             return null;
         }
+
+        if (!connectionOutService.isLocal(table.getNode())) {
+            logger.debug("This Controller is not the master for connector : "+table);
+            return null;
+        }
+
         return filter.readNodeTable(containerName, table, cached);
     }
 
@@ -182,11 +221,20 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
             return null;
         }
 
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return null;
+        }
+
         return filter.readAllNodeTable(containerName, node, cached);
     }
 
     @Override
     public void nodeFlowStatisticsUpdated(Node node, List<FlowOnNode> flowStatsList) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return;
+        }
         for (IPluginOutReadService service : pluginOutReadServices) {
             service.nodeFlowStatisticsUpdated(node, flowStatsList);
         }
@@ -194,6 +242,10 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
 
     @Override
     public void nodeConnectorStatisticsUpdated(Node node, List<NodeConnectorStatistics> ncStatsList) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return;
+        }
         for (IPluginOutReadService service : pluginOutReadServices) {
             service.nodeConnectorStatisticsUpdated(node, ncStatsList);
         }
@@ -201,6 +253,10 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
 
     @Override
     public void nodeTableStatisticsUpdated(Node node, List<NodeTableStatistics> tableStatsList) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return;
+        }
         for (IPluginOutReadService service : pluginOutReadServices) {
             service.nodeTableStatisticsUpdated(node, tableStatsList);
         }
@@ -208,8 +264,22 @@ public class ReadService implements IPluginInReadService, IReadFilterInternalLis
 
     @Override
     public void nodeDescriptionStatisticsUpdated(Node node, NodeDescription nodeDescription) {
+        if (!connectionOutService.isLocal(node)) {
+            logger.debug("This Controller is not the master for node : " + node);
+            return;
+        }
         for (IPluginOutReadService service : pluginOutReadServices) {
             service.descriptionStatisticsUpdated(node, nodeDescription);
         }
     }
+
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
 }

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadServiceFilter.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadServiceFilter.java
index 2c8708f20ee46295972a6b18b280df436f009dfc..1ab89b34cb7c875b65e1dee5be409ebbbc5e1378 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadServiceFilter.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/ReadServiceFilter.java
@@ -26,6 +26,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
 import org.opendaylight.controller.sal.action.Action;
 import org.opendaylight.controller.sal.action.ActionType;
 import org.opendaylight.controller.sal.action.Output;
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.ContainerFlow;
 import org.opendaylight.controller.sal.core.IContainerListener;
 import org.opendaylight.controller.sal.core.Node;
@@ -157,6 +158,17 @@ public class ReadServiceFilter implements IReadServiceFilter, IContainerListener
         this.statsMgr = null;
     }
 
+    IPluginOutConnectionService connectionPluginOutService;
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionPluginOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionPluginOutService == s) {
+            connectionPluginOutService = null;
+        }
+    }
+
     @Override
     public FlowOnNode readFlow(String container, Node node, Flow flow, boolean cached) {
 

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
index 1d7b4263e7b191cc13e83666130464442b66b5f2..a0f48100bdc812e4d1cda33c482bfc51b2f63587 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServiceShim.java
@@ -35,6 +35,7 @@ import org.osgi.framework.FrameworkUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Bandwidth;
 import org.opendaylight.controller.sal.core.Config;
 import org.opendaylight.controller.sal.core.ContainerFlow;
@@ -398,6 +399,17 @@ public class TopologyServiceShim implements IDiscoveryListener,
         }
     }
 
+    IPluginOutConnectionService connectionPluginOutService;
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionPluginOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionPluginOutService == s) {
+            connectionPluginOutService = null;
+        }
+    }
+
     private void removeNodeConnector(String container,
             NodeConnector nodeConnector) {
         List<TopoEdgeUpdate> teuList = new ArrayList<TopoEdgeUpdate>();

diff --git a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServices.java b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServices.java
index 68572fb65086ef4c7901dc4179caf03e57c12233..d77e513f3bd52f880b311a513705f88316a61c5a 100644 (file)
--- a/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServices.java
+++ b/opendaylight/protocol_plugins/openflow/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/TopologyServices.java
@@ -16,6 +16,7 @@ import org.opendaylight.controller.protocol_plugin.openflow.ITopologyServiceShim
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.opendaylight.controller.sal.connection.IPluginOutConnectionService;
 import org.opendaylight.controller.sal.core.Edge;
 import org.opendaylight.controller.sal.topology.IPluginInTopologyService;
 import org.opendaylight.controller.sal.topology.IPluginOutTopologyService;
@@ -27,6 +28,7 @@ public class TopologyServices implements ITopologyServiceShimListener,
             .getLogger(TopologyServices.class);
     private IPluginOutTopologyService salTopoService = null;
     private IRefreshInternalProvider topoRefreshService = null;
+    private IPluginOutConnectionService connectionOutService;
     private String containerName;
 
     /**
@@ -150,4 +152,14 @@ public class TopologyServices implements ITopologyServiceShimListener,
             this.salTopoService.edgeUtilBackToNormal(edge);
         }
     }
+
+    void setIPluginOutConnectionService(IPluginOutConnectionService s) {
+        connectionOutService = s;
+    }
+
+    void unsetIPluginOutConnectionService(IPluginOutConnectionService s) {
+        if (connectionOutService == s) {
+            connectionOutService = null;
+        }
+    }
 }