CORD-537 Added flow rules for vSG connectivity

- Added Q_IN_Q table
- Added flow rules for vSG connectivity
- Changed to listen port update event from Neutron to update vSG IPs

Change-Id: I227ba7a91e90ec0752481ebf623b4e848d585265

@@ -38,7 +38,6 @@ import org.onosproject.net.Host;
 import org.onosproject.net.HostId;
 import org.onosproject.net.HostId;
 import org.onosproject.net.HostLocation;
 import org.onosproject.net.HostLocation;
 import org.onosproject.net.Port;
 import org.onosproject.net.Port;
-import org.onosproject.net.SparseAnnotations;
 import org.onosproject.net.config.ConfigFactory;
 import org.onosproject.net.config.ConfigFactory;
 import org.onosproject.net.config.NetworkConfigEvent;
 import org.onosproject.net.config.NetworkConfigEvent;
 import org.onosproject.net.config.NetworkConfigListener;
 import org.onosproject.net.config.NetworkConfigListener;
@@ -137,10 +136,14 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
             };
             };
 
 
     private static final String DEFAULT_TUNNEL = "vxlan";
     private static final String DEFAULT_TUNNEL = "vxlan";
-    private static final Ip4Address DEFAULT_DNS = Ip4Address.valueOf("8.8.8.8");
     private static final String SERVICE_ID = "serviceId";
     private static final String SERVICE_ID = "serviceId";
-    private static final String LOCATION_IP = "locationIp";
     private static final String OPENSTACK_VM_ID = "openstackVmId";
     private static final String OPENSTACK_VM_ID = "openstackVmId";
+    private static final String OPENSTACK_PORT_ID = "openstackPortId";
+    private static final String DATA_PLANE_IP = "dataPlaneIp";
+    private static final String DATA_PLANE_INTF = "dataPlaneIntf";
+    private static final String S_TAG = "stag";
+
+    private static final Ip4Address DEFAULT_DNS = Ip4Address.valueOf("8.8.8.8");
 
 
     private final ExecutorService eventExecutor =
     private final ExecutorService eventExecutor =
             newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn", "event-handler"));
             newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn", "event-handler"));
@@ -263,18 +266,24 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
         }
         }
 
 
         Set<IpAddress> ip = Sets.newHashSet(vPort.fixedIps().values());
         Set<IpAddress> ip = Sets.newHashSet(vPort.fixedIps().values());
-        SparseAnnotations annotations = DefaultAnnotations.builder()
-                .set(OPENSTACK_VM_ID, vPort.deviceId())
+        DefaultAnnotations.Builder annotations = DefaultAnnotations.builder()
                 .set(SERVICE_ID, vPort.networkId())
                 .set(SERVICE_ID, vPort.networkId())
-                .set(LOCATION_IP, node.dpIp().ip().toString())
-                .build();
+                .set(OPENSTACK_VM_ID, vPort.deviceId())
+                .set(OPENSTACK_PORT_ID, vPort.id())
+                .set(DATA_PLANE_IP, node.dpIp().ip().toString())
+                .set(DATA_PLANE_INTF, node.dpIntf());
+
+        String serviceVlan = getServiceVlan(vPort);
+        if (serviceVlan != null) {
+            annotations.set(S_TAG, serviceVlan);
+        }
 
 
         HostDescription hostDesc = new DefaultHostDescription(
         HostDescription hostDesc = new DefaultHostDescription(
                 mac,
                 mac,
                 VlanId.NONE,
                 VlanId.NONE,
                 new HostLocation(connectPoint, System.currentTimeMillis()),
                 new HostLocation(connectPoint, System.currentTimeMillis()),
                 ip,
                 ip,
-                annotations);
+                annotations.build());
 
 
         hostProvider.hostDetected(hostId, hostDesc, false);
         hostProvider.hostDetected(hostId, hostDesc, false);
     }
     }
@@ -294,6 +303,20 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
         hostProvider.hostVanished(host.id());
         hostProvider.hostVanished(host.id());
     }
     }
 
 
+    @Override
+    public void updateVirtualSubscriberGateways(HostId vSgHostId, String serviceVlan,
+                                                Set<IpAddress> vSgIps) {
+        Host vSgVm = hostService.getHost(vSgHostId);
+
+        if (vSgVm == null || !vSgVm.annotations().value(S_TAG).equals(serviceVlan)) {
+            log.debug("Invalid vSG updates for {}", serviceVlan);
+            return;
+        }
+
+        log.info("Updates vSGs in {} with {}", vSgVm.id(), vSgIps.toString());
+        ruleInstaller.populateSubscriberGatewayRules(vSgVm, vSgIps);
+    }
+
     /**
     /**
      * Returns CordService by service ID.
      * Returns CordService by service ID.
      *
      *
@@ -357,10 +380,11 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
      * Returns IP address for tunneling for a given host.
      * Returns IP address for tunneling for a given host.
      *
      *
      * @param host host
      * @param host host
-     * @return ip address
+     * @return ip address, or null
      */
      */
     private IpAddress getTunnelIp(Host host) {
     private IpAddress getTunnelIp(Host host) {
-        return IpAddress.valueOf(host.annotations().value(LOCATION_IP));
+        String ip = host.annotations().value(DATA_PLANE_IP);
+        return ip == null ? null : IpAddress.valueOf(ip);
     }
     }
 
 
     /**
     /**
@@ -374,6 +398,22 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
     }
     }
 
 
     /**
     /**
+     * Returns s-tag from a given OpenStack port.
+     *
+     * @param vPort openstack port
+     * @return s-tag string
+     */
+    private String getServiceVlan(OpenstackPort vPort) {
+        checkNotNull(vPort);
+
+        if (vPort.name() != null && vPort.name().startsWith(S_TAG)) {
+            return vPort.name().split("-")[1];
+        } else {
+            return null;
+        }
+    }
+
+    /**
      * Returns hosts associated with a given OpenStack network.
      * Returns hosts associated with a given OpenStack network.
      *
      *
      * @param vNet openstack network
      * @param vNet openstack network
@@ -395,6 +435,30 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
     }
     }
 
 
     /**
     /**
+     * Returns public ip addresses of vSGs running inside a give vSG host.
+     *
+     * @param vSgHost vSG host
+     * @return set of ip address, or empty set
+     */
+    private Set<IpAddress> getSubscriberGatewayIps(Host vSgHost) {
+        String vPortId = vSgHost.annotations().value(OPENSTACK_PORT_ID);
+        String serviceVlan = vSgHost.annotations().value(S_TAG);
+
+        OpenstackPort vPort = openstackService.port(vPortId);
+        if (vPort == null) {
+            log.warn("Failed to get OpenStack port {} for VM {}", vPortId, vSgHost.id());
+            return Sets.newHashSet();
+        }
+
+        if (!serviceVlan.equals(getServiceVlan(vPort))) {
+            log.error("Host({}) s-tag does not match with vPort s-tag", vSgHost.id());
+            return Sets.newHashSet();
+        }
+
+        return vPort.allowedAddressPairs().keySet();
+    }
+
+    /**
      * Registers static DHCP lease for a given host.
      * Registers static DHCP lease for a given host.
      *
      *
      * @param host host
      * @param host host
@@ -452,8 +516,13 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
             arpProxy.sendGratuitousArp(service.serviceIp(), gatewayMac, Sets.newHashSet(host));
             arpProxy.sendGratuitousArp(service.serviceIp(), gatewayMac, Sets.newHashSet(host));
         }
         }
 
 
-        ruleInstaller.populateBasicConnectionRules(host, getTunnelIp(host), vNet);
         registerDhcpLease(host, service);
         registerDhcpLease(host, service);
+        ruleInstaller.populateBasicConnectionRules(host, getTunnelIp(host), vNet);
+
+        if (host.annotations().value(S_TAG) != null) {
+            log.debug("vSG VM detected {}", host.id());
+            ruleInstaller.populateSubscriberGatewayRules(host, getSubscriberGatewayIps(host));
+        }
     }
     }
 
 
     /**
     /**
@@ -468,7 +537,7 @@ public class CordVtn extends AbstractProvider implements CordVtnService, HostPro
         }
         }
 
 
         String vNetId = host.annotations().value(SERVICE_ID);
         String vNetId = host.annotations().value(SERVICE_ID);
-        OpenstackNetwork vNet = openstackService.network(host.annotations().value(SERVICE_ID));
+        OpenstackNetwork vNet = openstackService.network(vNetId);
         if (vNet == null) {
         if (vNet == null) {
             log.warn("Failed to get OpenStack network {} for VM {}({}).",
             log.warn("Failed to get OpenStack network {} for VM {}({}).",
                      vNetId,
                      vNetId,

@@ -15,7 +15,11 @@
  */
  */
 package org.onosproject.cordvtn;
 package org.onosproject.cordvtn;
 
 
+import org.onlab.packet.IpAddress;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.HostId;
+
+import java.util.Set;
 
 
 /**
 /**
  * Service for provisioning overlay virtual networks on compute nodes.
  * Service for provisioning overlay virtual networks on compute nodes.
@@ -57,4 +61,14 @@ public interface CordVtnService {
      * @param pServiceId id of the service which provide dependency
      * @param pServiceId id of the service which provide dependency
      */
      */
     void removeServiceDependency(CordServiceId tServiceId, CordServiceId pServiceId);
     void removeServiceDependency(CordServiceId tServiceId, CordServiceId pServiceId);
+
+    /**
+     * Updates virtual service gateways.
+     *
+     * @param vSgHost host id of vSG host
+     * @param serviceVlan service vlan id
+     * @param vSgIps set of ip address of vSGs running in this vSG host
+     */
+    void updateVirtualSubscriberGateways(HostId vSgHost, String serviceVlan,
+                                         Set<IpAddress> vSgIps);
 }
 }

@@ -220,7 +220,7 @@ public final class RemoteIpCommandUtil {
             return null;
             return null;
         }
         }
 
 
-        log.debug("Execute command {} to {}", command, session.getHost());
+        log.trace("Execute command {} to {}", command, session.getHost());
 
 
         try {
         try {
             Channel channel = session.openChannel("exec");
             Channel channel = session.openChannel("exec");

@@ -15,6 +15,13 @@
  */
  */
 package org.onosproject.cordvtn.rest;
 package org.onosproject.cordvtn.rest;
 
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Sets;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.MacAddress;
+import org.onosproject.cordvtn.CordVtnService;
+import org.onosproject.net.HostId;
 import org.onosproject.rest.AbstractWebResource;
 import org.onosproject.rest.AbstractWebResource;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
@@ -29,16 +36,29 @@ import javax.ws.rs.Produces;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response;
 import java.io.InputStream;
 import java.io.InputStream;
+import java.util.Set;
+
 
 
 /**
 /**
  * Dummy Neutron ML2 mechanism driver.
  * Dummy Neutron ML2 mechanism driver.
- * It just returns OK for ports resource requests.
+ * It just returns OK for ports resource requests except for the port update.
  */
  */
 @Path("ports")
 @Path("ports")
 public class NeutronMl2PortsWebResource extends AbstractWebResource {
 public class NeutronMl2PortsWebResource extends AbstractWebResource {
     protected final Logger log = LoggerFactory.getLogger(getClass());
     protected final Logger log = LoggerFactory.getLogger(getClass());
     private static final String PORTS_MESSAGE = "Received ports %s";
     private static final String PORTS_MESSAGE = "Received ports %s";
 
 
+    private static final String PORT = "port";
+    private static final String DEVICE_ID = "device_id";
+    private static final String NAME = "name";
+    private static final String MAC_ADDRESS = "mac_address";
+    private static final String ADDRESS_PAIRS = "allowed_address_pairs";
+    private static final String IP_ADDERSS = "ip_address";
+    private static final String STAG_PREFIX = "stag-";
+    private static final int STAG_BEGIN_INDEX = 5;
+
+    private final CordVtnService service = get(CordVtnService.class);
+
     @POST
     @POST
     @Consumes(MediaType.APPLICATION_JSON)
     @Consumes(MediaType.APPLICATION_JSON)
     @Produces(MediaType.APPLICATION_JSON)
     @Produces(MediaType.APPLICATION_JSON)
@@ -53,6 +73,35 @@ public class NeutronMl2PortsWebResource extends AbstractWebResource {
     @Produces(MediaType.APPLICATION_JSON)
     @Produces(MediaType.APPLICATION_JSON)
     public Response updatePorts(@PathParam("id") String id, InputStream input) {
     public Response updatePorts(@PathParam("id") String id, InputStream input) {
         log.debug(String.format(PORTS_MESSAGE, "update"));
         log.debug(String.format(PORTS_MESSAGE, "update"));
+
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            JsonNode jsonNode = mapper.readTree(input).get(PORT);
+            log.trace("{}", jsonNode.toString());
+
+            String deviceId = jsonNode.path(DEVICE_ID).asText();
+            String name = jsonNode.path(NAME).asText();
+            if (deviceId.isEmpty() || name.isEmpty() || !name.startsWith(STAG_PREFIX)) {
+                // ignore all updates other than allowed address pairs
+                return Response.status(Response.Status.OK).build();
+            }
+
+            // this is allowed address pairs updates
+            MacAddress mac = MacAddress.valueOf(jsonNode.path(MAC_ADDRESS).asText());
+            Set<IpAddress> vSgIps = Sets.newHashSet();
+            jsonNode.path(ADDRESS_PAIRS).forEach(addrPair -> {
+                IpAddress ip = IpAddress.valueOf(addrPair.path(IP_ADDERSS).asText());
+                vSgIps.add(ip);
+            });
+
+            service.updateVirtualSubscriberGateways(
+                    HostId.hostId(mac),
+                    name.substring(STAG_BEGIN_INDEX),
+                    vSgIps);
+        } catch (Exception e) {
+            return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
+        }
+
         return Response.status(Response.Status.OK).build();
         return Response.status(Response.Status.OK).build();
     }
     }