Merge into master from pull request #149:

Add a flow reconciliation pipeline (https://github.com/floodlight/floodlight/pull/149)

src/main/java/net/floodlightcontroller/devicemanager/internal/DeviceManagerImpl.java

@@ -55,6 +55,7 @@ import net.floodlightcontroller.devicemanager.IDeviceListener;
 import net.floodlightcontroller.devicemanager.SwitchPort;
 import net.floodlightcontroller.devicemanager.web.DeviceRoutable;
 import net.floodlightcontroller.flowcache.IFlowReconcileListener;
+import net.floodlightcontroller.flowcache.IFlowReconcileService;
 import net.floodlightcontroller.flowcache.OFMatchReconcile;
 import net.floodlightcontroller.packet.ARP;
 import net.floodlightcontroller.packet.DHCP;
@@ -96,6 +97,7 @@ public class DeviceManagerImpl implements
     protected IStorageSourceService storageSource;
     protected IRestApiService restApi;
     protected IThreadPoolService threadPool;
+    protected IFlowReconcileService flowReconcileMgr;
     
     /**
      * Time in milliseconds before entities will expire
@@ -618,6 +620,7 @@ public class DeviceManagerImpl implements
                 fmc.getServiceImpl(ITopologyService.class);
         this.restApi = fmc.getServiceImpl(IRestApiService.class);
         this.threadPool = fmc.getServiceImpl(IThreadPoolService.class);
+        this.flowReconcileMgr = fmc.getServiceImpl(IFlowReconcileService.class);
     }
     
     @Override
@@ -634,6 +637,7 @@ public class DeviceManagerImpl implements
         apComparator = new AttachmentPointComparator();
         
         floodlightProvider.addOFMessageListener(OFType.PACKET_IN, this);
+        flowReconcileMgr.addFlowReconcileListener(this);
         
         Runnable ecr = new Runnable() {
             @Override

src/main/java/net/floodlightcontroller/flowcache/FlowReconcileManager.java

+package net.floodlightcontroller.flowcache;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+
+import net.floodlightcontroller.core.module.FloodlightModuleContext;
+import net.floodlightcontroller.core.module.FloodlightModuleException;
+import net.floodlightcontroller.core.module.IFloodlightModule;
+import net.floodlightcontroller.core.module.IFloodlightService;
+import net.floodlightcontroller.core.util.ListenerDispatcher;
+import net.floodlightcontroller.flowcache.IFlowReconcileListener;
+import net.floodlightcontroller.flowcache.OFMatchReconcile;
+
+import org.openflow.protocol.OFType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * This class registers for various network events that may require flow 
+ * reconciliation. Examples include host-move, new attachment-point,
+ * switch connection etc. 
+ * 
+ * @author subrata
+ *
+ */
+public class FlowReconcileManager 
+        implements IFloodlightModule, IFlowReconcileService {
+
+    /** The logger. */
+    private static Logger logger =
+                        LoggerFactory.getLogger(FlowReconcileManager.class);
+
+    /**
+     * The list of flow reconcile listeners that have registered to get
+     * flow reconcile callbacks. Such callbacks are invoked, for example, when
+     * a switch with existing flow-mods joins this controller and those flows
+     * need to be reconciled with the current configuration of the controller.
+     */
+    protected ListenerDispatcher<OFType, IFlowReconcileListener> flowReconcileListeners;
+    
+    public OFMatchReconcile newOFMatchReconcile() {
+        return new OFMatchReconcile();
+    }
+
+    @Override
+    public synchronized void addFlowReconcileListener(IFlowReconcileListener listener) {
+        flowReconcileListeners.addListener(OFType.FLOW_MOD, listener);
+
+        if (logger.isDebugEnabled()) {
+            StringBuffer sb = new StringBuffer();
+            sb.append("FlowReconcileManager FlowMod Listeners: ");
+            for (IFlowReconcileListener l : flowReconcileListeners.getOrderedListeners()) {
+                sb.append(l.getName());
+                sb.append(",");
+            }
+            logger.debug(sb.toString());
+        }
+    }
+
+    @Override
+    public synchronized void removeFlowReconcileListener(IFlowReconcileListener listener) {
+        flowReconcileListeners.removeListener(listener);
+    }
+    
+    @Override
+    public synchronized void clearFlowReconcileListeners() {
+        flowReconcileListeners.clearListeners();
+    }
+    
+    /**
+     * Reconcile flow.
+     *
+     * @param ofmRcIn the ofm rc in
+     */
+    public void reconcileFlow(OFMatchReconcile ofmRcIn) {
+        if (logger.isTraceEnabled()) {
+            logger.trace("Reconciliating flow: {}", ofmRcIn.toString());
+        }
+        ArrayList<OFMatchReconcile> ofmRcList =
+                                            new ArrayList<OFMatchReconcile>();
+        ofmRcList.add(ofmRcIn);
+        // Run the flow through all the flow reconcile listeners
+        IFlowReconcileListener.Command retCmd;
+        for (IFlowReconcileListener flowReconciler : flowReconcileListeners.getOrderedListeners()) {
+            if (logger.isTraceEnabled()) {
+                logger.trace("Reconciliatng flow: call listener {}", flowReconciler.getName());
+            }
+            retCmd = flowReconciler.reconcileFlows(ofmRcList);
+            if (retCmd == IFlowReconcileListener.Command.STOP) {
+                break;
+            }
+        }
+    }
+    
+    // IFloodlightModule
+
+    @Override
+    public Collection<Class<? extends IFloodlightService>> getModuleServices() {
+        Collection<Class<? extends IFloodlightService>> l = 
+                new ArrayList<Class<? extends IFloodlightService>>();
+        l.add(IFlowReconcileService.class);
+        return l;
+    }
+
+    @Override
+    public Map<Class<? extends IFloodlightService>, IFloodlightService> 
+                                                            getServiceImpls() {
+    	Map<Class<? extends IFloodlightService>,
+    	IFloodlightService> m = 
+            new HashMap<Class<? extends IFloodlightService>,
+                IFloodlightService>();
+        m.put(IFlowReconcileService.class, this);
+        return m;
+    }
+
+    @Override
+    public Collection<Class<? extends IFloodlightService>> 
+                                                    getModuleDependencies() {
+        return null;
+    }
+
+    @Override
+    public void init(FloodlightModuleContext context)
+            throws FloodlightModuleException {
+        flowReconcileListeners = 
+                new ListenerDispatcher<OFType, IFlowReconcileListener>();
+    }
+
+    @Override
+    public void startUp(FloodlightModuleContext context) {
+    }
+}
+

src/main/java/net/floodlightcontroller/flowcache/IFlowReconcileService.java

+/**
+ * Provides Flow Reconcile service to other modules that need to reconcile
+ * flows.
+ */
+package net.floodlightcontroller.flowcache;
+
+import net.floodlightcontroller.core.module.IFloodlightService;
+
+public interface IFlowReconcileService extends IFloodlightService {
+    /**
+     * Add a flow reconcile listener
+     * @param listener The module that can reconcile flows
+     */
+    public void addFlowReconcileListener(IFlowReconcileListener listener);
+
+    /**
+     * Remove a flow reconcile listener
+     * @param listener The module that no longer reconcile flows
+     */
+    public void removeFlowReconcileListener(IFlowReconcileListener listener);
+    
+    /**
+     * Remove all flow reconcile listeners
+     */
+    public void clearFlowReconcileListeners();
+    
+    /**
+     * Reconcile flow. Returns false if no modified flow-mod need to be 
+     * programmed if cluster ID is providced then pnly flows in the given 
+     * cluster are reprogrammed
+     *
+     * @param ofmRcIn the ofm rc in
+     */
+    public void reconcileFlow(OFMatchReconcile ofmRcIn);
+
+}

src/main/java/net/floodlightcontroller/flowcache/OFMatchReconcile.java

@@ -25,7 +25,7 @@ public class OFMatchReconcile  {
 	     * for example when a host is moved or when a link goes down. */
 	    UPDATE_PATH,
 	    /* Flow is now in a different BVS */
-	    BVS_CHANGED,
+	    APP_INSTANCE_CHANGED,
 	    /* Delete the flow-mod - used to delete, for example, drop flow-mods
 	     * when the source and destination are in the same BVS after a 
 	     * configuration change */

src/main/resources/META-INF/services/net.floodlightcontroller.core.module.IFloodlightModule

@@ -4,6 +4,7 @@ net.floodlightcontroller.devicemanager.internal.DeviceManagerImpl
 net.floodlightcontroller.linkdiscovery.internal.LinkDiscoveryManager
 net.floodlightcontroller.topology.TopologyManager
 net.floodlightcontroller.forwarding.Forwarding
+net.floodlightcontroller.flowcache.FlowReconcileManager
 net.floodlightcontroller.core.OFMessageFilterManager
 net.floodlightcontroller.staticflowentry.StaticFlowEntryPusher
 net.floodlightcontroller.perfmon.PktInProcessingTime
@@ -16,4 +17,4 @@ net.floodlightcontroller.counter.CounterStore
 net.floodlightcontroller.counter.NullCounterStore
 net.floodlightcontroller.threadpool.ThreadPool
 net.floodlightcontroller.ui.web.StaticWebRoutable
-net.floodlightcontroller.virtualnetwork.forwarding.VirtualNetworkFilter
\ No newline at end of file
+net.floodlightcontroller.virtualnetwork.forwarding.VirtualNetworkFilter

src/test/java/net/floodlightcontroller/devicemanager/internal/DeviceManagerImplTest.java

@@ -48,6 +48,8 @@ import net.floodlightcontroller.devicemanager.IDevice;
 import net.floodlightcontroller.devicemanager.IEntityClass;
 import net.floodlightcontroller.devicemanager.SwitchPort;
 import net.floodlightcontroller.devicemanager.IDeviceService;
+import net.floodlightcontroller.flowcache.FlowReconcileManager;
+import net.floodlightcontroller.flowcache.IFlowReconcileService;
 import net.floodlightcontroller.packet.ARP;
 import net.floodlightcontroller.packet.Ethernet;
 import net.floodlightcontroller.packet.IPacket;
@@ -81,6 +83,7 @@ public class DeviceManagerImplTest extends FloodlightTestCase {
     MockFloodlightProvider mockFloodlightProvider;
     DeviceManagerImpl deviceManager;
     MemoryStorageSource storageSource;
+    FlowReconcileManager flowReconcileMgr;
     
     private IOFSwitch makeSwitchMock(long id) {
         IOFSwitch mockSwitch = createMock(IOFSwitch.class);
@@ -104,17 +107,21 @@ public class DeviceManagerImplTest extends FloodlightTestCase {
         fmc.addService(IThreadPoolService.class, tp);
         mockFloodlightProvider = getMockFloodlightProvider();
         deviceManager = new DeviceManagerImpl();
+        flowReconcileMgr = new FlowReconcileManager();
         fmc.addService(IDeviceService.class, deviceManager);
         storageSource = new MemoryStorageSource();
         fmc.addService(IStorageSourceService.class, storageSource);
         fmc.addService(IFloodlightProviderService.class, mockFloodlightProvider);
         fmc.addService(IRestApiService.class, restApi);
+        fmc.addService(IFlowReconcileService.class, flowReconcileMgr);
         tp.init(fmc);
         restApi.init(fmc);
         storageSource.init(fmc);
         deviceManager.init(fmc);
+        flowReconcileMgr.init(fmc);
         storageSource.startUp(fmc);
         deviceManager.startUp(fmc);
+        flowReconcileMgr.startUp(fmc);
         tp.startUp(fmc);
         
         IOFSwitch mockSwitch1 = makeSwitchMock(1L);

src/test/java/net/floodlightcontroller/flowcache/FlowReconcileMgrTest.java

+package net.floodlightcontroller.flowcache;
+
+import static org.easymock.EasyMock.*;
+
+import java.util.ArrayList;
+
+import net.floodlightcontroller.core.IListener.Command;
+import net.floodlightcontroller.core.module.FloodlightModuleContext;
+import net.floodlightcontroller.core.test.MockFloodlightProvider;
+import net.floodlightcontroller.flowcache.IFlowReconcileListener;
+import net.floodlightcontroller.flowcache.OFMatchReconcile;
+import net.floodlightcontroller.test.FloodlightTestCase;
+
+import org.easymock.EasyMock;
+import org.easymock.IAnswer;
+import org.junit.Before;
+import org.junit.Test;
+import org.openflow.protocol.OFStatisticsRequest;
+import org.openflow.protocol.OFType;
+
+/**
+ * The Class FlowCacheTest. Tests flow cache operations for
+ * (a) adding a flow to the flow cache when flow mod is programmed
+ * (b) Deleting a flow from flow cache when flow mod removal mesg is received
+ * (c) Reprogramming a flow based on flows queried from flow cache when a
+ *     device is moved.
+ * (d) Reprogramming a flow based on flows queried from flow cache when a
+ *     link goes down.
+ *
+ * @author subrata
+ */
+public class FlowReconcileMgrTest extends FloodlightTestCase {
+
+    protected MockFloodlightProvider mockFloodlightProvider;
+    protected FlowReconcileManager flowReconcileMgr;
+    
+    OFStatisticsRequest ofStatsRequest;
+
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+
+        FloodlightModuleContext fmc = new FloodlightModuleContext();
+        flowReconcileMgr = new FlowReconcileManager();
+        
+        flowReconcileMgr.init(fmc);
+    }
+
+    /** Verify pipeline listener registration and ordering
+     * 
+     * @throws Exception
+     */
+    @SuppressWarnings("unchecked")
+    @Test
+    public void testFlowReconcilePipeLine() throws Exception {
+        IFlowReconcileListener r1 = EasyMock.createNiceMock(IFlowReconcileListener.class);
+        IFlowReconcileListener r2 = EasyMock.createNiceMock(IFlowReconcileListener.class);
+        IFlowReconcileListener r3 = EasyMock.createNiceMock(IFlowReconcileListener.class);
+        
+        expect(r1.getName()).andReturn("r1").anyTimes();
+        expect(r2.getName()).andReturn("r2").anyTimes();
+        expect(r3.getName()).andReturn("r3").anyTimes();
+        
+        // Set the listeners' order: r1 -> r2 -> r3
+        expect(r1.isCallbackOrderingPrereq((OFType)anyObject(), (String)anyObject())).andReturn(false).anyTimes();
+        expect(r1.isCallbackOrderingPostreq((OFType)anyObject(), (String)anyObject())).andReturn(false).anyTimes();
+        expect(r2.isCallbackOrderingPrereq((OFType)anyObject(), eq("r1"))).andReturn(true).anyTimes();
+        expect(r2.isCallbackOrderingPrereq((OFType)anyObject(), eq("r3"))).andReturn(false).anyTimes();
+        expect(r2.isCallbackOrderingPostreq((OFType)anyObject(), eq("r1"))).andReturn(false).anyTimes();
+        expect(r2.isCallbackOrderingPostreq((OFType)anyObject(), eq("r3"))).andReturn(true).anyTimes();
+        expect(r3.isCallbackOrderingPrereq((OFType)anyObject(), eq("r1"))).andReturn(false).anyTimes();
+        expect(r3.isCallbackOrderingPrereq((OFType)anyObject(), eq("r2"))).andReturn(true).anyTimes();
+        expect(r3.isCallbackOrderingPostreq((OFType)anyObject(), (String)anyObject())).andReturn(false).anyTimes();
+        
+        expect(r1.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).
+                  andThrow(new RuntimeException("This is NOT an error! We are testing exception catching."));
+        
+        replay(r1, r2, r3);
+        flowReconcileMgr.clearFlowReconcileListeners();
+        flowReconcileMgr.addFlowReconcileListener(r1);
+        flowReconcileMgr.addFlowReconcileListener(r2);
+        flowReconcileMgr.addFlowReconcileListener(r3);
+        
+        OFMatchReconcile ofmRcIn = flowReconcileMgr.newOFMatchReconcile();
+        try {
+            flowReconcileMgr.reconcileFlow(ofmRcIn);
+        } catch (RuntimeException e) {
+            assertEquals(e.getMessage().startsWith("This is NOT an error!"), true);
+        }
+        verify(r1, r2, r3);
+
+        // verify STOP works
+        reset(r1, r2, r3);
+        expect(r1.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.STOP).times(1);
+        expect(r2.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject()));
+        expectLastCall().andAnswer(new IAnswer<Object>() {
+            public Object answer() {
+                fail("Unexpected call");
+                return Command.STOP;
+            }
+        }).anyTimes();
+        
+        replay(r1, r2, r3);
+        flowReconcileMgr.reconcileFlow(ofmRcIn);
+        verify(r1, r2, r3);
+        
+        // verify CONTINUE works
+        reset(r1, r2, r3);
+        expect(r1.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.CONTINUE).times(1);
+        expect(r2.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.STOP).times(1);
+        expect(r3.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject()));
+        expectLastCall().andAnswer(new IAnswer<Object>() {
+            public Object answer() {
+                fail("Unexpected call");
+                return Command.STOP;
+            }
+        }).anyTimes();
+        
+        replay(r1, r2, r3);
+        flowReconcileMgr.reconcileFlow(ofmRcIn);
+        verify(r1, r2, r3);
+        
+        // verify CONTINUE works
+        reset(r1, r2, r3);
+        expect(r1.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.CONTINUE).times(1);
+        expect(r2.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.CONTINUE).times(1);
+        expect(r3.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.STOP).times(1);
+        replay(r1, r2, r3);
+        flowReconcileMgr.reconcileFlow(ofmRcIn);
+        verify(r1, r2, r3);
+        
+        // Verify removeFlowReconcileListener
+        flowReconcileMgr.removeFlowReconcileListener(r1);
+        reset(r1, r2, r3);
+        expect(r1.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject()));
+        expectLastCall().andAnswer(new IAnswer<Object>() {
+            public Object answer() {
+                fail("Unexpected call to a listener that is removed from the chain.");
+                return Command.STOP;
+            }
+        }).anyTimes();
+        expect(r2.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.CONTINUE).times(1);
+        expect(r3.reconcileFlows((ArrayList<OFMatchReconcile>)anyObject())).andReturn(Command.STOP).times(1);
+        replay(r1, r2, r3);
+        flowReconcileMgr.reconcileFlow(ofmRcIn);
+        verify(r1, r2, r3);
+    }
+}
\ No newline at end of file

src/test/java/net/floodlightcontroller/forwarding/ForwardingTest.java

@@ -54,6 +54,8 @@ import net.floodlightcontroller.routing.Route;
 import net.floodlightcontroller.test.FloodlightTestCase;
 import net.floodlightcontroller.threadpool.IThreadPoolService;
 import net.floodlightcontroller.topology.ITopologyService;
+import net.floodlightcontroller.flowcache.FlowReconcileManager;
+import net.floodlightcontroller.flowcache.IFlowReconcileService;
 import net.floodlightcontroller.forwarding.Forwarding;
 
 import org.easymock.Capture;
@@ -76,6 +78,7 @@ public class ForwardingTest extends FloodlightTestCase {
     protected MockDeviceManager deviceManager;
     protected IRoutingService routingEngine;
     protected Forwarding forwarding;
+    protected FlowReconcileManager flowReconcileMgr;
     protected ITopologyService topology;
     protected MockThreadPoolService threadPool;
     protected IOFSwitch sw1, sw2;
@@ -97,8 +100,10 @@ public class ForwardingTest extends FloodlightTestCase {
         forwarding = getForwarding();
         threadPool = new MockThreadPoolService();
         deviceManager = new MockDeviceManager();
+        flowReconcileMgr = new FlowReconcileManager();
         routingEngine = createMock(IRoutingService.class);
         topology = createMock(ITopologyService.class);
+        
 
         FloodlightModuleContext fmc = new FloodlightModuleContext();
         fmc.addService(IFloodlightProviderService.class, 
@@ -108,13 +113,16 @@ public class ForwardingTest extends FloodlightTestCase {
         fmc.addService(IRoutingService.class, routingEngine);
         fmc.addService(ICounterStoreService.class, new CounterStore());
         fmc.addService(IDeviceService.class, deviceManager);
+        fmc.addService(IFlowReconcileService.class, flowReconcileMgr);
 
         threadPool.init(fmc);
         forwarding.init(fmc);
         deviceManager.init(fmc);
+        flowReconcileMgr.init(fmc);
         threadPool.startUp(fmc);
         deviceManager.startUp(fmc);
         forwarding.startUp(fmc);
+        flowReconcileMgr.startUp(fmc);
         
         // Mock switches
         sw1 = EasyMock.createNiceMock(IOFSwitch.class);