Changes to vRouter to accomodate ofdpa and softrouter pipelines.

Adding arp-spa to flow from vRouter to distinguish between multiple untagged
interfaces with the same macAddress.

Change-Id: Ifd6e00f70c538c780c0f5728d9ba960a4c70b1db

@@ -22,6 +22,7 @@ import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.onlab.packet.EthType;
+import org.onlab.packet.VlanId;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
 import org.onosproject.incubator.net.intf.Interface;
@@ -40,8 +41,10 @@ import org.onosproject.net.flow.DefaultTrafficTreatment;
 import org.onosproject.net.flow.TrafficSelector;
 import org.onosproject.net.flow.TrafficTreatment;
 import org.onosproject.net.flowobjective.DefaultForwardingObjective;
+import org.onosproject.net.flowobjective.DefaultNextObjective;
 import org.onosproject.net.flowobjective.FlowObjectiveService;
 import org.onosproject.net.flowobjective.ForwardingObjective;
+import org.onosproject.net.flowobjective.NextObjective;
 import org.onosproject.net.host.InterfaceIpAddress;
 import org.onosproject.routing.RoutingService;
 import org.onosproject.routing.config.RouterConfig;
@@ -110,7 +113,7 @@ public class ControlPlaneRedirectManager {
                 routingAppId, RoutingService.ROUTER_CONFIG_CLASS);
 
         if (config == null) {
-            log.info("Router config not available");
+            log.warn("Router config not available");
             return;
         }
 
@@ -145,6 +148,23 @@ public class ControlPlaneRedirectManager {
         PortNumber controlPlanePort = controlPlaneConnectPoint.port();
 
         for (InterfaceIpAddress ip : intf.ipAddresses()) {
+            // create nextObjectives for forwarding to this interface and the
+            // controlPlaneConnectPoint
+            int cpNextId, intfNextId;
+            if (intf.vlan() == VlanId.NONE) {
+                cpNextId = createNextObjective(deviceId, controlPlanePort,
+                               VlanId.vlanId(SingleSwitchFibInstaller.ASSIGNED_VLAN),
+                               true);
+                intfNextId = createNextObjective(deviceId, intf.connectPoint().port(),
+                               VlanId.vlanId(SingleSwitchFibInstaller.ASSIGNED_VLAN),
+                               true);
+            } else {
+                cpNextId = createNextObjective(deviceId, controlPlanePort,
+                                               intf.vlan(), false);
+                intfNextId = createNextObjective(deviceId, intf.connectPoint().port(),
+                                                 intf.vlan(), false);
+            }
+
             // IPv4 to router
             TrafficSelector toSelector = DefaultTrafficSelector.builder()
                     .matchInPort(intf.connectPoint().port())
@@ -154,12 +174,8 @@ public class ControlPlaneRedirectManager {
                     .matchIPDst(ip.ipAddress().toIpPrefix())
                     .build();
 
-            TrafficTreatment toTreatment = DefaultTrafficTreatment.builder()
-                    .setOutput(controlPlanePort)
-                    .build();
-
             flowObjectiveService.forward(deviceId,
-                    buildForwardingObjective(toSelector, toTreatment, true));
+                    buildForwardingObjective(toSelector, null, cpNextId, true));
 
             // IPv4 from router
             TrafficSelector fromSelector = DefaultTrafficSelector.builder()
@@ -170,12 +186,8 @@ public class ControlPlaneRedirectManager {
                     .matchIPSrc(ip.ipAddress().toIpPrefix())
                     .build();
 
-            TrafficTreatment intfTreatment = DefaultTrafficTreatment.builder()
-                    .setOutput(intf.connectPoint().port())
-                    .build();
-
             flowObjectiveService.forward(deviceId,
-                    buildForwardingObjective(fromSelector, intfTreatment, true));
+                    buildForwardingObjective(fromSelector, null, intfNextId, true));
 
             // ARP to router
             toSelector = DefaultTrafficSelector.builder()
@@ -184,13 +196,12 @@ public class ControlPlaneRedirectManager {
                     .matchVlanId(intf.vlan())
                     .build();
 
-            toTreatment = DefaultTrafficTreatment.builder()
-                    .setOutput(controlPlanePort)
+            TrafficTreatment puntTreatment = DefaultTrafficTreatment.builder()
                     .punt()
                     .build();
 
             flowObjectiveService.forward(deviceId,
-                    buildForwardingObjective(toSelector, toTreatment, true));
+                    buildForwardingObjective(toSelector, puntTreatment, cpNextId, true));
 
             // ARP from router
             fromSelector = DefaultTrafficSelector.builder()
@@ -198,15 +209,11 @@ public class ControlPlaneRedirectManager {
                     .matchEthSrc(intf.mac())
                     .matchVlanId(intf.vlan())
                     .matchEthType(EthType.EtherType.ARP.ethType().toShort())
-                    .build();
-
-            intfTreatment = DefaultTrafficTreatment.builder()
-                    .setOutput(intf.connectPoint().port())
-                    .punt()
+                    .matchArpSpa(ip.ipAddress().getIp4Address())
                     .build();
 
             flowObjectiveService.forward(deviceId,
-                    buildForwardingObjective(fromSelector, intfTreatment, true));
+            buildForwardingObjective(fromSelector, puntTreatment, intfNextId, true));
         }
     }
 
@@ -219,33 +226,85 @@ public class ControlPlaneRedirectManager {
                 .matchIPProtocol((byte) OSPF_IP_PROTO)
                 .build();
 
-        TrafficTreatment toTreatment = DefaultTrafficTreatment.builder()
-                .setOutput(controlPlaneConnectPoint.port())
-                .build();
-
+        // create nextObjectives for forwarding to the controlPlaneConnectPoint
+        DeviceId deviceId = controlPlaneConnectPoint.deviceId();
+        PortNumber controlPlanePort = controlPlaneConnectPoint.port();
+        int cpNextId;
+        if (intf.vlan() == VlanId.NONE) {
+            cpNextId = createNextObjective(deviceId, controlPlanePort,
+                           VlanId.vlanId(SingleSwitchFibInstaller.ASSIGNED_VLAN),
+                           true);
+        } else {
+            cpNextId = createNextObjective(deviceId, controlPlanePort,
+                                           intf.vlan(), false);
+        }
+        log.debug("ospf flows intf:{} nextid:{}", intf, cpNextId);
         flowObjectiveService.forward(controlPlaneConnectPoint.deviceId(),
-                buildForwardingObjective(toSelector, toTreatment, ospfEnabled));
+                buildForwardingObjective(toSelector, null, cpNextId, ospfEnabled));
     }
 
     /**
-     * Builds a forwarding objective from the given selector and treatment.
+     * Creates a next objective for forwarding to a port. Handles metadata for
+     * some pipelines that require vlan information for egress port.
+     *
+     * @param deviceId the device on which the next objective is being created
+     * @param portNumber the egress port
+     * @param vlanId vlan information for egress port
+     * @param popVlan if vlan tag should be popped or not
+     * @return nextId of the next objective created
+     */
+    private int createNextObjective(DeviceId deviceId, PortNumber portNumber,
+                                    VlanId vlanId, boolean popVlan) {
+        int nextId = flowObjectiveService.allocateNextId();
+        NextObjective.Builder nextObjBuilder = DefaultNextObjective
+                .builder().withId(nextId)
+                .withType(NextObjective.Type.SIMPLE)
+                .fromApp(appId);
+
+        TrafficTreatment.Builder ttBuilder = DefaultTrafficTreatment.builder();
+        if (popVlan) {
+            ttBuilder.popVlan();
+        }
+        ttBuilder.setOutput(portNumber);
+
+        // setup metadata to pass to nextObjective - indicate the vlan on egress
+        // if needed by the switch pipeline.
+        TrafficSelector.Builder metabuilder = DefaultTrafficSelector.builder();
+        metabuilder.matchVlanId(vlanId);
+
+        nextObjBuilder.withMeta(metabuilder.build());
+        nextObjBuilder.addTreatment(ttBuilder.build());
+        log.debug("Submited next objective {} in device {} for port/vlan {}/{}",
+                nextId, deviceId, portNumber, vlanId);
+        flowObjectiveService.next(deviceId, nextObjBuilder.add());
+        return nextId;
+    }
+
+    /**
+     * Builds a forwarding objective from the given selector, treatment and nextId.
      *
      * @param selector selector
-     * @param treatment treatment
+     * @param treatment treatment to apply to packet, can be null
+     * @param nextId next objective to point to for forwarding packet
      * @param add true to create an add objective, false to create a remove
      *            objective
      * @return forwarding objective
      */
     private ForwardingObjective buildForwardingObjective(TrafficSelector selector,
                                                          TrafficTreatment treatment,
+                                                         int nextId,
                                                          boolean add) {
-
-        ForwardingObjective.Builder fobBuilder = DefaultForwardingObjective.builder()
-                .withSelector(selector)
-                .withTreatment(treatment)
-                .fromApp(appId)
-                .withPriority(PRIORITY)
-                .withFlag(ForwardingObjective.Flag.VERSATILE);
+        DefaultForwardingObjective.Builder fobBuilder = DefaultForwardingObjective.builder();
+        fobBuilder.withSelector(selector);
+        if (treatment != null) {
+            fobBuilder.withTreatment(treatment);
+        }
+        if (nextId != -1) {
+            fobBuilder.nextStep(nextId);
+        }
+        fobBuilder.fromApp(appId)
+            .withPriority(PRIORITY)
+            .withFlag(ForwardingObjective.Flag.VERSATILE);
 
         return add ? fobBuilder.add() : fobBuilder.remove();
     }

@@ -34,6 +34,7 @@ import org.onosproject.core.ApplicationId;
 import org.onosproject.core.CoreService;
 import org.onosproject.incubator.net.intf.Interface;
 import org.onosproject.incubator.net.intf.InterfaceService;
+import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.config.NetworkConfigEvent;
 import org.onosproject.net.config.NetworkConfigListener;
@@ -80,6 +81,8 @@ public class SingleSwitchFibInstaller {
     private static final int PRIORITY_OFFSET = 100;
     private static final int PRIORITY_MULTIPLIER = 5;
 
+    public static final short ASSIGNED_VLAN = 4094;
+
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected CoreService coreService;
 
@@ -103,7 +106,8 @@ public class SingleSwitchFibInstaller {
     // Device id of data-plane switch - should be learned from config
     private DeviceId deviceId;
 
-    private ApplicationId appId;
+    private ConnectPoint controlPlaneConnectPoint;
+
     private ApplicationId routerAppId;
 
     // Reference count for how many times a next hop is used by a route
@@ -121,11 +125,8 @@ public class SingleSwitchFibInstaller {
 
     @Activate
     protected void activate() {
-        // TODO why are there two of the same app ID?
         routerAppId = coreService.registerApplication(RoutingService.ROUTER_APP_ID);
 
-        appId = coreService.getAppId(RoutingService.ROUTER_APP_ID);
-
         deviceListener = new InternalDeviceListener();
         deviceService.addListener(deviceListener);
 
@@ -156,6 +157,8 @@ public class SingleSwitchFibInstaller {
             log.info("Router config not available");
             return;
         }
+        controlPlaneConnectPoint = routerConfig.getControlPlaneConnectPoint();
+        log.info("Control Plane Connect Point: {}", controlPlaneConnectPoint);
 
         deviceId = routerConfig.getControlPlaneConnectPoint().deviceId();
 
@@ -231,7 +234,7 @@ public class SingleSwitchFibInstaller {
         int priority = prefix.prefixLength() * PRIORITY_MULTIPLIER + PRIORITY_OFFSET;
 
         ForwardingObjective.Builder fwdBuilder = DefaultForwardingObjective.builder()
-                .fromApp(appId)
+                .fromApp(routerAppId)
                 .makePermanent()
                 .withSelector(selector)
                 .withPriority(priority)
@@ -266,23 +269,30 @@ public class SingleSwitchFibInstaller {
                     .setEthSrc(egressIntf.mac())
                     .setEthDst(nextHop.mac());
 
+            TrafficSelector.Builder metabuilder = null;
             if (!egressIntf.vlan().equals(VlanId.NONE)) {
                 treatment.pushVlan()
                         .setVlanId(egressIntf.vlan())
                         .setVlanPcp((byte) 0);
+            } else {
+                // untagged outgoing port may require internal vlan in some pipelines
+                metabuilder = DefaultTrafficSelector.builder();
+                metabuilder.matchVlanId(VlanId.vlanId(ASSIGNED_VLAN));
             }
 
             treatment.setOutput(egressIntf.connectPoint().port());
 
             int nextId = flowObjectiveService.allocateNextId();
-
-            NextObjective nextObjective = DefaultNextObjective.builder()
+            NextObjective.Builder nextBuilder = DefaultNextObjective.builder()
                     .withId(nextId)
                     .addTreatment(treatment.build())
                     .withType(NextObjective.Type.SIMPLE)
-                    .fromApp(appId)
-                    .add(); // TODO add callbacks
+                    .fromApp(routerAppId);
+            if (metabuilder != null) {
+                nextBuilder.withMeta(metabuilder.build());
+            }
 
+            NextObjective nextObjective = nextBuilder.add(); // TODO add callbacks
             flowObjectiveService.next(deviceId, nextObjective);
 
             nextHops.put(nextHop.ip(), nextId);
@@ -329,33 +339,48 @@ public class SingleSwitchFibInstaller {
             }
 
             FilteringObjective.Builder fob = DefaultFilteringObjective.builder();
+            // first add filter for the interface
             fob.withKey(Criteria.matchInPort(intf.connectPoint().port()))
-                    .addCondition(Criteria.matchEthDst(intf.mac()))
-                    .addCondition(Criteria.matchVlanId(intf.vlan()));
-
+                .addCondition(Criteria.matchEthDst(intf.mac()))
+                .addCondition(Criteria.matchVlanId(intf.vlan()));
             fob.withPriority(PRIORITY_OFFSET);
+            if (intf.vlan() == VlanId.NONE) {
+                TrafficTreatment tt = DefaultTrafficTreatment.builder()
+                        .pushVlan().setVlanId(VlanId.vlanId(ASSIGNED_VLAN)).build();
+                fob.withMeta(tt);
+            }
 
-            fob.permit().fromApp(appId);
-            flowObjectiveService.filter(
-                    deviceId,
-                    fob.add(new ObjectiveContext() {
-                        @Override
-                        public void onSuccess(Objective objective) {
-                            log.info("Successfully installed interface based "
-                                    + "filtering objectives for intf {}", intf);
-                        }
-
-                        @Override
-                        public void onError(Objective objective,
-                                            ObjectiveError error) {
-                            log.error("Failed to install interface filters for intf {}: {}",
-                                    intf, error);
-                            // TODO something more than just logging
-                        }
-                    }));
+            fob.permit().fromApp(routerAppId);
+            sendFilteringObjective(install, fob, intf);
+            if (controlPlaneConnectPoint != null) {
+                // then add the same mac/vlan filters for control-plane connect point
+                fob.withKey(Criteria.matchInPort(controlPlaneConnectPoint.port()));
+                sendFilteringObjective(install, fob, intf);
+            }
         }
     }
 
+    private void sendFilteringObjective(boolean install, FilteringObjective.Builder fob,
+                                        Interface intf) {
+        flowObjectiveService.filter(
+            deviceId,
+            fob.add(new ObjectiveContext() {
+                @Override
+                public void onSuccess(Objective objective) {
+                    log.info("Successfully installed interface based "
+                            + "filtering objectives for intf {}", intf);
+                }
+
+                @Override
+                public void onError(Objective objective,
+                                    ObjectiveError error) {
+                    log.error("Failed to install interface filters for intf {}: {}",
+                              intf, error);
+                    // TODO something more than just logging
+                }
+            }));
+    }
+
     private class InternalFibListener implements FibListener {
 
         @Override

@@ -387,11 +387,12 @@ public class FlowObjectiveManager implements FlowObjectiveService {
                 Set<PendingNext> pending = pendingForwards.remove(event.subject());
 
                 if (pending == null) {
-                    log.debug("Nothing pending for this obj event");
+                    log.warn("Nothing pending for this obj event {}", event);
                     return;
                 }
 
-                log.debug("Processing pending forwarding objectives {}", pending.size());
+                log.debug("Processing {} pending forwarding objectives for nextId {}",
+                         pending.size(), event.subject());
                 pending.forEach(p -> getDevicePipeliner(p.deviceId())
                                 .forward(p.forwardingObjective()));
             }

@@ -584,34 +584,59 @@ public class OFDPA2Pipeline extends AbstractHandlerBehaviour implements Pipeline
                     + "nextId or Treatment", fwd.selector(), fwd.appId());
             return Collections.emptySet();
         }
+
         // XXX driver does not currently do type checking as per Tables 65-67 in
         // OFDPA 2.0 spec. The only allowed treatment is a punt to the controller.
-        if (fwd.treatment() != null &&
-                fwd.treatment().allInstructions().size() == 1 &&
-                fwd.treatment().allInstructions().get(0).type() == Instruction.Type.OUTPUT) {
-            OutputInstruction o = (OutputInstruction) fwd.treatment().allInstructions().get(0);
-            if (o.port() == PortNumber.CONTROLLER) {
-                FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
-                        .fromApp(fwd.appId())
-                        .withPriority(fwd.priority())
-                        .forDevice(deviceId)
-                        .withSelector(fwd.selector())
-                        .withTreatment(fwd.treatment())
-                        .makePermanent()
-                        .forTable(ACL_TABLE);
-                return Collections.singletonList(ruleBuilder.build());
-            } else {
-                log.warn("Only allowed treatments in versatile forwarding "
-                        + "objectives are punts to the controller");
-                return Collections.emptySet();
+        TrafficTreatment.Builder ttBuilder = DefaultTrafficTreatment.builder();
+        if (fwd.treatment() != null) {
+            for (Instruction ins : fwd.treatment().allInstructions()) {
+                if (ins instanceof OutputInstruction) {
+                    OutputInstruction o = (OutputInstruction) ins;
+                    if (o.port() == PortNumber.CONTROLLER) {
+                        ttBuilder.add(o);
+                    } else {
+                        log.warn("Only allowed treatments in versatile forwarding "
+                                + "objectives are punts to the controller");
+                    }
+                } else {
+                    log.warn("Cannot process instruction in versatile fwd {}", ins);
+                }
             }
         }
-
         if (fwd.nextId() != null) {
-            // XXX overide case
-            log.warn("versatile objective --> next Id not yet implemeted");
-        }
-        return Collections.emptySet();
+            // overide case
+            NextGroup next = getGroupForNextObjective(fwd.nextId());
+            List<Deque<GroupKey>> gkeys = appKryo.deserialize(next.data());
+            // we only need the top level group's key to point the flow to it
+            Group group = groupService.getGroup(deviceId, gkeys.get(0).peekFirst());
+            if (group == null) {
+                log.warn("Group with key:{} for next-id:{} not found in dev:{}",
+                         gkeys.get(0).peekFirst(), fwd.nextId(), deviceId);
+                fail(fwd, ObjectiveError.GROUPMISSING);
+                return Collections.emptySet();
+            }
+            ttBuilder.deferred().group(group.id());
+        }
+       // ensure that match does not include vlan = NONE as OF-DPA does not
+       // match untagged packets this way in the ACL table.
+       TrafficSelector.Builder selBuilder = DefaultTrafficSelector.builder();
+       for (Criterion criterion : fwd.selector().criteria()) {
+           if (criterion instanceof VlanIdCriterion &&
+               ((VlanIdCriterion) criterion).vlanId() == VlanId.NONE) {
+               continue;
+           }
+           selBuilder.add(criterion);
+       }
+
+        FlowRule.Builder ruleBuilder = DefaultFlowRule.builder()
+                .fromApp(fwd.appId())
+                .withPriority(fwd.priority())
+                .forDevice(deviceId)
+                .withSelector(selBuilder.build())
+                .withTreatment(ttBuilder.build())
+                .makePermanent()
+                .forTable(ACL_TABLE);
+        return Collections.singletonList(ruleBuilder.build());
     }
 
     /**

@@ -42,6 +42,8 @@ import org.onosproject.net.flow.criteria.EthTypeCriterion;
 import org.onosproject.net.flow.criteria.IPCriterion;
 import org.onosproject.net.flow.criteria.PortCriterion;
 import org.onosproject.net.flow.criteria.VlanIdCriterion;
+import org.onosproject.net.flow.instructions.Instruction;
+import org.onosproject.net.flow.instructions.Instructions.OutputInstruction;
 import org.onosproject.net.flowobjective.FilteringObjective;
 import org.onosproject.net.flowobjective.FlowObjectiveStore;
 import org.onosproject.net.flowobjective.ForwardingObjective;
@@ -50,6 +52,7 @@ import org.onosproject.net.flowobjective.Objective;
 import org.onosproject.net.flowobjective.ObjectiveError;
 import org.onosproject.store.serializers.KryoNamespaces;
 import org.slf4j.Logger;
+import static org.onlab.util.Tools.delay;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -314,19 +317,50 @@ public class SoftRouterPipeline extends AbstractHandlerBehaviour implements Pipe
     }
 
     /**
-     * SoftRouter has a single versatile table - the filter table. All versatile
-     * flow rules must include the filtering rules.
+     * SoftRouter has a single versatile table - the filter table.
+     * This table can be used to filter entries that reach the next table (FIB table).
+     * It can also be used to punt packets to the controller and/or bypass
+     * the FIB table to forward out of a port.
      *
      * @param fwd The forwarding objective of type versatile
      * @return  A collection of flow rules meant to be delivered to the flowrule
      *          subsystem. May return empty collection in case of failures.
      */
     private Collection<FlowRule> processVersatile(ForwardingObjective fwd) {
+        log.debug("Received versatile fwd: to next:{}", fwd.nextId());
         Collection<FlowRule> flowrules = new ArrayList<>();
+        if (fwd.nextId() == null && fwd.treatment() == null) {
+            log.error("Forwarding objective {} from {} must contain "
+                    + "nextId or Treatment", fwd.selector(), fwd.appId());
+            return Collections.emptySet();
+        }
+        TrafficTreatment.Builder ttBuilder = DefaultTrafficTreatment.builder();
+        if (fwd.treatment() != null) {
+            fwd.treatment().immediate().forEach(ins -> ttBuilder.add(ins));
+        }
+        //convert nextId to flow actions
+        if (fwd.nextId() != null) {
+            // only acceptable value is output to port
+            NextGroup next = flowObjectiveStore.getNextGroup(fwd.nextId());
+            if (next == null) {
+                log.error("next-id {} does not exist in store", fwd.nextId());
+                return Collections.emptySet();
+            }
+            TrafficTreatment nt = appKryo.deserialize(next.data());
+            if (nt == null)  {
+                log.error("Error in deserializing next-id {}", fwd.nextId());
+                return Collections.emptySet();
+            }
+            for (Instruction ins : nt.allInstructions()) {
+                if (ins instanceof OutputInstruction) {
+                    ttBuilder.add(ins);
+                }
+            }
+        }
 
         FlowRule rule = DefaultFlowRule.builder()
                 .withSelector(fwd.selector())
-                .withTreatment(fwd.treatment())
+                .withTreatment(ttBuilder.build())
                 .makePermanent()
                 .forDevice(deviceId)
                 .fromApp(fwd.appId())
@@ -350,7 +384,7 @@ public class SoftRouterPipeline extends AbstractHandlerBehaviour implements Pipe
      *
      */
     private Collection<FlowRule> processSpecific(ForwardingObjective fwd) {
-        log.debug("Processing specific forwarding objective");
+        log.debug("Processing specific forwarding objective to next:{}", fwd.nextId());
         TrafficSelector selector = fwd.selector();
         EthTypeCriterion ethType =
                 (EthTypeCriterion) selector.getCriterion(Criterion.Type.ETH_TYPE);
@@ -411,6 +445,9 @@ public class SoftRouterPipeline extends AbstractHandlerBehaviour implements Pipe
      */
     private void processSimpleNextObjective(NextObjective nextObj) {
         // Simple next objective has a single treatment (not a collection)
+        log.debug("Received nextObj {}", nextObj.id());
+        // delay processing to emulate group creation
+        delay(50);
         TrafficTreatment treatment = nextObj.next().iterator().next();
         flowObjectiveStore.putNextGroup(nextObj.id(),
                                         new DummyGroup(treatment));