flowReconcile pipeline

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
@@ -615,6 +617,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
@@ -631,6 +634,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.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() {
+        return null;
+    }
+
+    @Override
+    public Map<Class<? extends IFloodlightService>, IFloodlightService> 
+                                                            getServiceImpls() {
+        return null;
+    }
+
+    @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/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