pom modifications to appify vtn
renaming package
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtn.java b/src/main/java/org/opencord/cordvtn/impl/CordVtn.java
new file mode 100644
index 0000000..f007a36
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtn.java
@@ -0,0 +1,410 @@
+/*
+ * Copyright 2015-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.Ip4Prefix;
+import org.opencord.cordvtn.api.CordVtnNode;
+import org.opencord.cordvtn.api.CordVtnService;
+import org.opencord.cordvtn.api.Instance;
+import org.onosproject.core.DefaultGroupId;
+import org.onosproject.core.GroupId;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.Host;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.flow.DefaultFlowRule;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.flow.instructions.ExtensionTreatment;
+import org.onosproject.net.group.DefaultGroupDescription;
+import org.onosproject.net.group.DefaultGroupKey;
+import org.onosproject.net.group.Group;
+import org.onosproject.net.group.GroupBucket;
+import org.onosproject.net.group.GroupBuckets;
+import org.onosproject.net.group.GroupDescription;
+import org.onosproject.net.group.GroupKey;
+import org.onosproject.net.group.GroupService;
+import org.onosproject.net.host.HostEvent;
+import org.onosproject.net.host.HostListener;
+import org.onosproject.xosclient.api.VtnService;
+import org.onosproject.xosclient.api.VtnServiceId;
+import org.slf4j.Logger;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.opencord.cordvtn.impl.CordVtnPipeline.*;
+import static org.onosproject.net.group.DefaultGroupBucket.createSelectGroupBucket;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Provisions service dependency capabilities between network services.
+ */
+@Component(immediate = true)
+@Service
+public class CordVtn extends CordVtnInstanceHandler implements CordVtnService {
+
+ protected final Logger log = getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected GroupService groupService;
+
+ @Activate
+ protected void activate() {
+ eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn", "event-handler"));
+ hostListener = new InternalHostListener();
+ super.activate();
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ super.deactivate();
+ }
+
+ @Override
+ public void createServiceDependency(VtnServiceId tServiceId, VtnServiceId pServiceId,
+ boolean isBidirectional) {
+ VtnService tService = getVtnService(tServiceId);
+ VtnService pService = getVtnService(pServiceId);
+
+ if (tService == null || pService == null) {
+ log.error("Failed to create dependency between {} and {}",
+ tServiceId, pServiceId);
+ return;
+ }
+
+ log.info("Created dependency between {} and {}", tService.name(), pService.name());
+ serviceDependencyRules(tService, pService, isBidirectional, true);
+ }
+
+ @Override
+ public void removeServiceDependency(VtnServiceId tServiceId, VtnServiceId pServiceId) {
+ VtnService tService = getVtnService(tServiceId);
+ VtnService pService = getVtnService(pServiceId);
+
+ if (tService == null || pService == null) {
+ log.error("Failed to remove dependency between {} and {}",
+ tServiceId, pServiceId);
+ return;
+ }
+
+ log.info("Removed dependency between {} and {}", tService.name(), pService.name());
+ serviceDependencyRules(tService, pService, true, false);
+ }
+
+ @Override
+ public void instanceDetected(Instance instance) {
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ return;
+ }
+
+ // TODO get bidirectional information from XOS once XOS supports
+ service.tenantServices().stream().forEach(
+ tServiceId -> createServiceDependency(tServiceId, service.id(), true));
+ service.providerServices().stream().forEach(
+ pServiceId -> createServiceDependency(service.id(), pServiceId, true));
+
+ updateProviderServiceInstances(service);
+ }
+
+ @Override
+ public void instanceRemoved(Instance instance) {
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ return;
+ }
+
+ if (!service.providerServices().isEmpty()) {
+ removeInstanceFromTenantService(instance, service);
+ }
+ if (!service.tenantServices().isEmpty()) {
+ updateProviderServiceInstances(service);
+ }
+ }
+
+ private void updateProviderServiceInstances(VtnService service) {
+ GroupKey groupKey = getGroupKey(service.id());
+
+ Set<DeviceId> devices = nodeManager.completeNodes().stream()
+ .map(CordVtnNode::intBrId)
+ .collect(Collectors.toSet());
+
+ for (DeviceId deviceId : devices) {
+ Group group = groupService.getGroup(deviceId, groupKey);
+ if (group == null) {
+ log.trace("No group exists for service {} in {}", service.id(), deviceId);
+ continue;
+ }
+
+ List<GroupBucket> oldBuckets = group.buckets().buckets();
+ List<GroupBucket> newBuckets = getServiceGroupBuckets(
+ deviceId, service.vni(), getInstances(service.id())).buckets();
+
+ if (oldBuckets.equals(newBuckets)) {
+ continue;
+ }
+
+ List<GroupBucket> bucketsToRemove = Lists.newArrayList(oldBuckets);
+ bucketsToRemove.removeAll(newBuckets);
+ if (!bucketsToRemove.isEmpty()) {
+ groupService.removeBucketsFromGroup(
+ deviceId,
+ groupKey,
+ new GroupBuckets(bucketsToRemove),
+ groupKey, appId);
+ }
+
+ List<GroupBucket> bucketsToAdd = Lists.newArrayList(newBuckets);
+ bucketsToAdd.removeAll(oldBuckets);
+ if (!bucketsToAdd.isEmpty()) {
+ groupService.addBucketsToGroup(
+ deviceId,
+ groupKey,
+ new GroupBuckets(bucketsToAdd),
+ groupKey, appId);
+ }
+ }
+ }
+
+ private void removeInstanceFromTenantService(Instance instance, VtnService service) {
+ service.providerServices().stream().forEach(pServiceId -> {
+ Map<DeviceId, Set<PortNumber>> inPorts = Maps.newHashMap();
+ Map<DeviceId, GroupId> outGroups = Maps.newHashMap();
+
+ inPorts.put(instance.deviceId(), Sets.newHashSet(instance.portNumber()));
+ outGroups.put(instance.deviceId(), getGroupId(pServiceId, instance.deviceId()));
+
+ inServiceRule(inPorts, outGroups, false);
+ });
+ }
+
+ private void serviceDependencyRules(VtnService tService, VtnService pService,
+ boolean isBidirectional, boolean install) {
+ Map<DeviceId, GroupId> outGroups = Maps.newHashMap();
+ Map<DeviceId, Set<PortNumber>> inPorts = Maps.newHashMap();
+
+ nodeManager.completeNodes().stream().forEach(node -> {
+ DeviceId deviceId = node.intBrId();
+ GroupId groupId = createServiceGroup(deviceId, pService);
+ outGroups.put(deviceId, groupId);
+
+ Set<PortNumber> tServiceInstances = getInstances(tService.id())
+ .stream()
+ .filter(instance -> instance.deviceId().equals(deviceId))
+ .map(Instance::portNumber)
+ .collect(Collectors.toSet());
+ inPorts.put(deviceId, tServiceInstances);
+ });
+
+ Ip4Prefix srcRange = tService.subnet().getIp4Prefix();
+ Ip4Prefix dstRange = pService.subnet().getIp4Prefix();
+
+ indirectAccessRule(srcRange, pService.serviceIp().getIp4Address(), outGroups, install);
+ directAccessRule(srcRange, dstRange, install);
+ if (isBidirectional) {
+ directAccessRule(dstRange, srcRange, install);
+ }
+ inServiceRule(inPorts, outGroups, install);
+ }
+
+ private void indirectAccessRule(Ip4Prefix srcRange, Ip4Address serviceIp,
+ Map<DeviceId, GroupId> outGroups, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPSrc(srcRange)
+ .matchIPDst(serviceIp.toIpPrefix())
+ .build();
+
+ for (Map.Entry<DeviceId, GroupId> outGroup : outGroups.entrySet()) {
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .group(outGroup.getValue())
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(outGroup.getKey())
+ .forTable(TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+ }
+
+ private void directAccessRule(Ip4Prefix srcRange, Ip4Prefix dstRange, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPSrc(srcRange)
+ .matchIPDst(dstRange)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .transition(TABLE_DST_IP)
+ .build();
+
+ nodeManager.completeNodes().stream().forEach(node -> {
+ DeviceId deviceId = node.intBrId();
+ FlowRule flowRuleDirect = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_DEFAULT)
+ .forDevice(deviceId)
+ .forTable(TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRuleDirect);
+ });
+ }
+
+ private void inServiceRule(Map<DeviceId, Set<PortNumber>> inPorts,
+ Map<DeviceId, GroupId> outGroups, boolean install) {
+ for (Map.Entry<DeviceId, Set<PortNumber>> entry : inPorts.entrySet()) {
+ Set<PortNumber> ports = entry.getValue();
+ DeviceId deviceId = entry.getKey();
+
+ GroupId groupId = outGroups.get(deviceId);
+ if (groupId == null) {
+ continue;
+ }
+
+ ports.stream().forEach(port -> {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(port)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .group(groupId)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_DEFAULT)
+ .forDevice(deviceId)
+ .forTable(TABLE_IN_SERVICE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ });
+ }
+ }
+
+ private GroupId getGroupId(VtnServiceId serviceId, DeviceId deviceId) {
+ return new DefaultGroupId(Objects.hash(serviceId, deviceId));
+ }
+
+ private GroupKey getGroupKey(VtnServiceId serviceId) {
+ return new DefaultGroupKey(serviceId.id().getBytes());
+ }
+
+ private GroupId createServiceGroup(DeviceId deviceId, VtnService service) {
+ GroupKey groupKey = getGroupKey(service.id());
+ Group group = groupService.getGroup(deviceId, groupKey);
+ GroupId groupId = getGroupId(service.id(), deviceId);
+
+ if (group != null) {
+ log.debug("Group {} is already exist in {}", service.id(), deviceId);
+ return groupId;
+ }
+
+ GroupBuckets buckets = getServiceGroupBuckets(
+ deviceId, service.vni(), getInstances(service.id()));
+ GroupDescription groupDescription = new DefaultGroupDescription(
+ deviceId,
+ GroupDescription.Type.SELECT,
+ buckets,
+ groupKey,
+ groupId.id(),
+ appId);
+
+ groupService.addGroup(groupDescription);
+ return groupId;
+ }
+
+ private GroupBuckets getServiceGroupBuckets(DeviceId deviceId, long tunnelId,
+ Set<Instance> instances) {
+ List<GroupBucket> buckets = Lists.newArrayList();
+ instances.stream().forEach(instance -> {
+ Ip4Address tunnelIp = nodeManager.dpIp(instance.deviceId()).getIp4Address();
+ TrafficTreatment.Builder tBuilder = DefaultTrafficTreatment.builder();
+
+ if (deviceId.equals(instance.deviceId())) {
+ tBuilder.setEthDst(instance.mac())
+ .setOutput(instance.portNumber());
+ } else {
+ ExtensionTreatment tunnelDst =
+ pipeline.tunnelDstTreatment(deviceId, tunnelIp);
+ tBuilder.setEthDst(instance.mac())
+ .extension(tunnelDst, deviceId)
+ .setTunnelId(tunnelId)
+ .setOutput(nodeManager.tunnelPort(instance.deviceId()));
+ }
+ buckets.add(createSelectGroupBucket(tBuilder.build()));
+ });
+ return new GroupBuckets(buckets);
+ }
+
+ private class InternalHostListener implements HostListener {
+
+ @Override
+ public void event(HostEvent event) {
+ Host host = event.subject();
+ if (!mastershipService.isLocalMaster(host.location().deviceId())) {
+ // do not allow to proceed without mastership
+ return;
+ }
+
+ Instance instance = Instance.of(host);
+ switch (event.type()) {
+ case HOST_UPDATED:
+ case HOST_ADDED:
+ eventExecutor.execute(() -> instanceDetected(instance));
+ break;
+ case HOST_REMOVED:
+ eventExecutor.execute(() -> instanceRemoved(instance));
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtnArpProxy.java b/src/main/java/org/opencord/cordvtn/impl/CordVtnArpProxy.java
new file mode 100644
index 0000000..6d29a7d
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtnArpProxy.java
@@ -0,0 +1,244 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import com.google.common.collect.Maps;
+import org.onlab.packet.ARP;
+import org.onlab.packet.EthType;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.MacAddress;
+import org.opencord.cordvtn.api.Instance;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.net.Host;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.host.HostService;
+import org.onosproject.net.packet.DefaultOutboundPacket;
+import org.onosproject.net.packet.PacketContext;
+import org.onosproject.net.packet.PacketPriority;
+import org.onosproject.net.packet.PacketService;
+import org.slf4j.Logger;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Handles ARP requests for virtual network service IPs.
+ */
+public class CordVtnArpProxy {
+ protected final Logger log = getLogger(getClass());
+
+ private final ApplicationId appId;
+ private final PacketService packetService;
+ private final HostService hostService;
+
+ private final Map<Ip4Address, MacAddress> gateways = Maps.newConcurrentMap();
+
+ /**
+ * Default constructor.
+ *
+ * @param appId application id
+ * @param packetService packet service
+ * @param hostService host service reference
+ */
+ public CordVtnArpProxy(ApplicationId appId, PacketService packetService, HostService hostService) {
+ this.appId = appId;
+ this.packetService = packetService;
+ this.hostService = hostService;
+ }
+
+ /**
+ * Requests ARP packet.
+ */
+ public void requestPacket() {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(EthType.EtherType.ARP.ethType().toShort())
+ .build();
+
+ packetService.requestPackets(selector,
+ PacketPriority.CONTROL,
+ appId,
+ Optional.empty());
+ }
+
+ /**
+ * Cancels ARP packet.
+ */
+ public void cancelPacket() {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(EthType.EtherType.ARP.ethType().toShort())
+ .build();
+
+ packetService.cancelPackets(selector,
+ PacketPriority.CONTROL,
+ appId,
+ Optional.empty());
+ }
+
+ /**
+ * Adds a given gateway IP and MAC address to this ARP proxy.
+ *
+ * @param gatewayIp gateway ip address
+ * @param gatewayMac gateway mac address
+ */
+ public void addGateway(IpAddress gatewayIp, MacAddress gatewayMac) {
+ checkNotNull(gatewayIp);
+ checkNotNull(gatewayMac);
+ gateways.put(gatewayIp.getIp4Address(), gatewayMac);
+ }
+
+ /**
+ * Removes a given service IP address from this ARP proxy.
+ *
+ * @param gatewayIp gateway ip address
+ */
+ public void removeGateway(IpAddress gatewayIp) {
+ checkNotNull(gatewayIp);
+ gateways.remove(gatewayIp.getIp4Address());
+ }
+
+ /**
+ * Emits ARP reply with fake MAC address for a given ARP request.
+ * It only handles requests for the registered service IPs, and the other
+ * requests can be handled by other ARP handlers like openstackSwitching or
+ * proxyArp, for example.
+ *
+ * @param context packet context
+ * @param ethPacket ethernet packet
+ */
+ public void processArpPacket(PacketContext context, Ethernet ethPacket) {
+ ARP arpPacket = (ARP) ethPacket.getPayload();
+ if (arpPacket.getOpCode() != ARP.OP_REQUEST) {
+ return;
+ }
+
+ Ip4Address targetIp = Ip4Address.valueOf(arpPacket.getTargetProtocolAddress());
+
+ MacAddress gatewayMac = gateways.get(targetIp);
+ MacAddress replyMac = gatewayMac != null ? gatewayMac : getMacFromHostService(targetIp);
+
+ if (replyMac.equals(MacAddress.NONE)) {
+ log.debug("Failed to find MAC for {}", targetIp.toString());
+ context.block();
+ return;
+ }
+
+ log.trace("Send ARP reply for {} with {}", targetIp.toString(), replyMac.toString());
+ Ethernet ethReply = ARP.buildArpReply(
+ targetIp,
+ replyMac,
+ ethPacket);
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(context.inPacket().receivedFrom().port())
+ .build();
+
+ packetService.emit(new DefaultOutboundPacket(
+ context.inPacket().receivedFrom().deviceId(),
+ treatment,
+ ByteBuffer.wrap(ethReply.serialize())));
+
+ context.block();
+ }
+
+ /**
+ * Emits gratuitous ARP when a gateway mac address has been changed.
+ *
+ * @param gatewayIp gateway ip address to update MAC
+ * @param instances set of instances to send gratuitous ARP packet
+ */
+ public void sendGratuitousArpForGateway(IpAddress gatewayIp, Set<Instance> instances) {
+ MacAddress gatewayMac = gateways.get(gatewayIp.getIp4Address());
+ if (gatewayMac == null) {
+ log.debug("Gateway {} is not registered to ARP proxy", gatewayIp.toString());
+ return;
+ }
+
+ Ethernet ethArp = buildGratuitousArp(gatewayIp.getIp4Address(), gatewayMac);
+ instances.stream().forEach(instance -> {
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(instance.portNumber())
+ .build();
+
+ packetService.emit(new DefaultOutboundPacket(
+ instance.deviceId(),
+ treatment,
+ ByteBuffer.wrap(ethArp.serialize())));
+ });
+ }
+
+ /**
+ * Builds gratuitous ARP packet with a given IP and MAC address.
+ *
+ * @param ip ip address for TPA and SPA
+ * @param mac new mac address
+ * @return ethernet packet
+ */
+ private Ethernet buildGratuitousArp(IpAddress ip, MacAddress mac) {
+ Ethernet eth = new Ethernet();
+
+ eth.setEtherType(Ethernet.TYPE_ARP);
+ eth.setSourceMACAddress(mac);
+ eth.setDestinationMACAddress(MacAddress.BROADCAST);
+
+ ARP arp = new ARP();
+ arp.setOpCode(ARP.OP_REQUEST);
+ arp.setHardwareType(ARP.HW_TYPE_ETHERNET);
+ arp.setHardwareAddressLength((byte) Ethernet.DATALAYER_ADDRESS_LENGTH);
+ arp.setProtocolType(ARP.PROTO_TYPE_IP);
+ arp.setProtocolAddressLength((byte) Ip4Address.BYTE_LENGTH);
+
+ arp.setSenderHardwareAddress(mac.toBytes());
+ arp.setTargetHardwareAddress(MacAddress.BROADCAST.toBytes());
+ arp.setSenderProtocolAddress(ip.getIp4Address().toOctets());
+ arp.setTargetProtocolAddress(ip.getIp4Address().toOctets());
+
+ eth.setPayload(arp);
+ return eth;
+ }
+
+ /**
+ * Returns MAC address of a host with a given target IP address by asking to
+ * host service. It does not support overlapping IP.
+ *
+ * @param targetIp target ip
+ * @return mac address, or NONE mac address if it fails to find the mac
+ */
+ private MacAddress getMacFromHostService(IpAddress targetIp) {
+ checkNotNull(targetIp);
+
+ Host host = hostService.getHostsByIp(targetIp)
+ .stream()
+ .findFirst()
+ .orElse(null);
+
+ if (host != null) {
+ log.trace("Found MAC from host service for {}", targetIp.toString());
+ return host.mac();
+ } else {
+ return MacAddress.NONE;
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceHandler.java b/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceHandler.java
new file mode 100644
index 0000000..242d94c
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceHandler.java
@@ -0,0 +1,535 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.Ip4Prefix;
+import org.opencord.cordvtn.api.CordVtnConfig;
+import org.opencord.cordvtn.api.CordVtnNode;
+import org.opencord.cordvtn.api.CordVtnService;
+import org.opencord.cordvtn.api.Instance;
+import org.opencord.cordvtn.api.InstanceHandler;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.Host;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.config.NetworkConfigEvent;
+import org.onosproject.net.config.NetworkConfigListener;
+import org.onosproject.net.config.NetworkConfigRegistry;
+import org.onosproject.net.flow.DefaultFlowRule;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.flow.instructions.ExtensionTreatment;
+import org.onosproject.net.host.HostEvent;
+import org.onosproject.net.host.HostListener;
+import org.onosproject.net.host.HostService;
+import org.onosproject.xosclient.api.OpenStackAccess;
+import org.onosproject.xosclient.api.VtnService;
+import org.onosproject.xosclient.api.VtnServiceApi;
+import org.onosproject.xosclient.api.VtnServiceId;
+import org.onosproject.xosclient.api.XosAccess;
+import org.onosproject.xosclient.api.XosClientService;
+import org.slf4j.Logger;
+
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.xosclient.api.VtnService.NetworkType.MANAGEMENT;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Provides default virtual network connectivity for service instances.
+ */
+@Component(immediate = true)
+public abstract class CordVtnInstanceHandler implements InstanceHandler {
+
+ protected final Logger log = getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected MastershipService mastershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected NetworkConfigRegistry configRegistry;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostService hostService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected XosClientService xosClient;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CordVtnNodeManager nodeManager;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CordVtnPipeline pipeline;
+
+ protected static final String OPENSTACK_ACCESS_ERROR = "OpenStack access is not configured";
+ protected static final String XOS_ACCESS_ERROR = "XOS access is not configured";
+
+ protected XosAccess xosAccess = null;
+ protected OpenStackAccess osAccess = null;
+ protected ApplicationId appId;
+ protected VtnService.ServiceType serviceType;
+ protected ExecutorService eventExecutor;
+
+ protected HostListener hostListener = new InternalHostListener();
+ protected NetworkConfigListener configListener = new InternalConfigListener();
+
+ protected void activate() {
+ // sub class should set service type and event executor in its activate method
+ appId = coreService.registerApplication(CordVtnService.CORDVTN_APP_ID);
+
+ hostService.addListener(hostListener);
+ configRegistry.addListener(configListener);
+
+ log.info("Started");
+ }
+
+ protected void deactivate() {
+ hostService.removeListener(hostListener);
+ configRegistry.removeListener(configListener);
+ eventExecutor.shutdown();
+
+ log.info("Stopped");
+ }
+
+ @Override
+ public void instanceDetected(Instance instance) {
+ log.info("Instance is detected {}", instance);
+
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ log.warn("Failed to get VtnService for {}", instance);
+ return;
+ }
+
+ if (service.networkType().equals(MANAGEMENT)) {
+ managementNetworkRules(instance, service, true);
+ }
+
+ defaultConnectionRules(instance, service, true);
+ }
+
+ @Override
+ public void instanceRemoved(Instance instance) {
+ log.info("Instance is removed {}", instance);
+
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ log.warn("Failed to get VtnService for {}", instance);
+ return;
+ }
+
+ if (service.networkType().equals(MANAGEMENT)) {
+ managementNetworkRules(instance, service, false);
+ }
+
+ // TODO check if any stale management network rules are
+ defaultConnectionRules(instance, service, false);
+ }
+
+ protected VtnService getVtnService(VtnServiceId serviceId) {
+ checkNotNull(osAccess, OPENSTACK_ACCESS_ERROR);
+ checkNotNull(xosAccess, XOS_ACCESS_ERROR);
+
+ // TODO remove openstack access when XOS provides all information
+ VtnServiceApi serviceApi = xosClient.getClient(xosAccess).vtnService();
+ VtnService service = serviceApi.service(serviceId, osAccess);
+ if (service == null) {
+ log.warn("Failed to get VtnService for {}", serviceId);
+ }
+ return service;
+ }
+
+ protected Set<Instance> getInstances(VtnServiceId serviceId) {
+ return StreamSupport.stream(hostService.getHosts().spliterator(), false)
+ .filter(host -> Objects.equals(
+ serviceId.id(),
+ host.annotations().value(Instance.SERVICE_ID)))
+ .map(Instance::of)
+ .collect(Collectors.toSet());
+ }
+
+ private void defaultConnectionRules(Instance instance, VtnService service, boolean install) {
+ long vni = service.vni();
+ Ip4Prefix serviceIpRange = service.subnet().getIp4Prefix();
+
+ inPortRule(instance, install);
+ dstIpRule(instance, vni, install);
+ tunnelInRule(instance, vni, install);
+
+ if (install) {
+ directAccessRule(serviceIpRange, serviceIpRange, true);
+ serviceIsolationRule(serviceIpRange, true);
+ } else if (getInstances(service.id()).isEmpty()) {
+ directAccessRule(serviceIpRange, serviceIpRange, false);
+ serviceIsolationRule(serviceIpRange, false);
+ }
+ }
+
+ private void managementNetworkRules(Instance instance, VtnService service, boolean install) {
+
+ managementPerInstanceRule(instance, install);
+ if (install) {
+ managementBaseRule(instance, service, true);
+ } else if (!hostService.getConnectedHosts(instance.deviceId()).stream()
+ .filter(host -> Instance.of(host).serviceId().equals(service.id()))
+ .findAny()
+ .isPresent()) {
+ managementBaseRule(instance, service, false);
+ }
+ }
+
+ private void managementBaseRule(Instance instance, VtnService service, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_ARP)
+ .matchArpTpa(service.serviceIp().getIp4Address())
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_MANAGEMENT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(PortNumber.LOCAL)
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(service.subnet())
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .transition(CordVtnPipeline.TABLE_DST_IP)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_MANAGEMENT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(service.serviceIp().toIpPrefix())
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_MANAGEMENT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+
+ private void managementPerInstanceRule(Instance instance, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(PortNumber.LOCAL)
+ .matchEthType(Ethernet.TYPE_ARP)
+ .matchArpTpa(instance.ipAddress().getIp4Address())
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(instance.portNumber())
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_MANAGEMENT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+
+ private void inPortRule(Instance instance, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(instance.portNumber())
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPSrc(instance.ipAddress().toIpPrefix())
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .transition(CordVtnPipeline.TABLE_ACCESS_TYPE)
+ .build();
+
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_IN_PORT)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(instance.portNumber())
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .transition(CordVtnPipeline.TABLE_IN_SERVICE)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_LOW)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_IN_PORT)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+
+ private void dstIpRule(Instance instance, long vni, boolean install) {
+ Ip4Address tunnelIp = nodeManager.dpIp(instance.deviceId()).getIp4Address();
+
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(instance.ipAddress().toIpPrefix())
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setEthDst(instance.mac())
+ .setOutput(instance.portNumber())
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_DST_IP)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ for (CordVtnNode node : nodeManager.completeNodes()) {
+ if (node.intBrId().equals(instance.deviceId())) {
+ continue;
+ }
+
+ ExtensionTreatment tunnelDst = pipeline.tunnelDstTreatment(node.intBrId(), tunnelIp);
+ if (tunnelDst == null) {
+ continue;
+ }
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setEthDst(instance.mac())
+ .setTunnelId(vni)
+ .extension(tunnelDst, node.intBrId())
+ .setOutput(nodeManager.tunnelPort(node.intBrId()))
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(node.intBrId())
+ .forTable(CordVtnPipeline.TABLE_DST_IP)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+ }
+
+ private void tunnelInRule(Instance instance, long vni, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchTunnelId(vni)
+ .matchEthDst(instance.mac())
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(instance.portNumber())
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(instance.deviceId())
+ .forTable(CordVtnPipeline.TABLE_TUNNEL_IN)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+
+ private void directAccessRule(Ip4Prefix srcRange, Ip4Prefix dstRange, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPSrc(srcRange)
+ .matchIPDst(dstRange)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .transition(CordVtnPipeline.TABLE_DST_IP)
+ .build();
+
+
+ nodeManager.completeNodes().stream().forEach(node -> {
+ FlowRule flowRuleDirect = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(node.intBrId())
+ .forTable(CordVtnPipeline.TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRuleDirect);
+ });
+ }
+
+ private void serviceIsolationRule(Ip4Prefix dstRange, boolean install) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(dstRange)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .drop()
+ .build();
+
+ nodeManager.completeNodes().stream().forEach(node -> {
+ FlowRule flowRuleDirect = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_LOW)
+ .forDevice(node.intBrId())
+ .forTable(CordVtnPipeline.TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRuleDirect);
+ });
+ }
+
+ protected void readConfiguration() {
+ CordVtnConfig config = configRegistry.getConfig(appId, CordVtnConfig.class);
+ if (config == null) {
+ log.debug("No configuration found");
+ return;
+ }
+ osAccess = config.openstackAccess();
+ xosAccess = config.xosAccess();
+ }
+
+ public class InternalHostListener implements HostListener {
+
+ @Override
+ public void event(HostEvent event) {
+ Host host = event.subject();
+ if (!mastershipService.isLocalMaster(host.location().deviceId())) {
+ // do not allow to proceed without mastership
+ return;
+ }
+
+ Instance instance = Instance.of(host);
+ if (!Objects.equals(instance.serviceType(), serviceType)) {
+ // not my service instance, do nothing
+ return;
+ }
+
+ switch (event.type()) {
+ case HOST_UPDATED:
+ case HOST_ADDED:
+ eventExecutor.execute(() -> instanceDetected(instance));
+ break;
+ case HOST_REMOVED:
+ eventExecutor.execute(() -> instanceRemoved(instance));
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ public class InternalConfigListener implements NetworkConfigListener {
+
+ @Override
+ public void event(NetworkConfigEvent event) {
+ if (!event.configClass().equals(CordVtnConfig.class)) {
+ return;
+ }
+
+ switch (event.type()) {
+ case CONFIG_ADDED:
+ case CONFIG_UPDATED:
+ readConfiguration();
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceManager.java b/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceManager.java
new file mode 100644
index 0000000..6acca13
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtnInstanceManager.java
@@ -0,0 +1,438 @@
+/*
+ * Copyright 2015-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import com.google.common.collect.Sets;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.VlanId;
+import org.opencord.cordvtn.api.CordVtnConfig;
+import org.opencord.cordvtn.api.CordVtnService;
+import org.opencord.cordvtn.api.Instance;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.dhcp.DhcpService;
+import org.onosproject.dhcp.IpAssignment;
+import org.onosproject.mastership.MastershipService;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.Host;
+import org.onosproject.net.HostId;
+import org.onosproject.net.HostLocation;
+import org.onosproject.net.Port;
+import org.onosproject.net.config.ConfigFactory;
+import org.onosproject.net.config.NetworkConfigEvent;
+import org.onosproject.net.config.NetworkConfigListener;
+import org.onosproject.net.config.NetworkConfigRegistry;
+import org.onosproject.net.config.basics.SubjectFactories;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.host.DefaultHostDescription;
+import org.onosproject.net.host.HostDescription;
+import org.onosproject.net.host.HostEvent;
+import org.onosproject.net.host.HostListener;
+import org.onosproject.net.host.HostProvider;
+import org.onosproject.net.host.HostProviderRegistry;
+import org.onosproject.net.host.HostProviderService;
+import org.onosproject.net.host.HostService;
+import org.onosproject.net.packet.PacketContext;
+import org.onosproject.net.packet.PacketProcessor;
+import org.onosproject.net.packet.PacketService;
+import org.onosproject.net.provider.AbstractProvider;
+import org.onosproject.net.provider.ProviderId;
+import org.onosproject.xosclient.api.OpenStackAccess;
+import org.onosproject.xosclient.api.VtnPort;
+import org.onosproject.xosclient.api.VtnPortApi;
+import org.onosproject.xosclient.api.VtnService;
+import org.onosproject.xosclient.api.VtnServiceApi;
+import org.onosproject.xosclient.api.VtnServiceId;
+import org.onosproject.xosclient.api.XosAccess;
+import org.onosproject.xosclient.api.XosClientService;
+import org.slf4j.Logger;
+
+import java.util.Date;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.dhcp.IpAssignment.AssignmentStatus.Option_RangeNotEnforced;
+import static org.onosproject.xosclient.api.VtnService.NetworkType.MANAGEMENT;
+import static org.onosproject.xosclient.api.VtnService.NetworkType.PRIVATE;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Adds or removes instances to network services.
+ */
+@Component(immediate = true)
+@Service(value = CordVtnInstanceManager.class)
+public class CordVtnInstanceManager extends AbstractProvider implements HostProvider {
+
+ protected final Logger log = getLogger(getClass());
+
+ private static final String XOS_ACCESS_ERROR = "XOS access is not configured";
+ private static final String OPENSTACK_ACCESS_ERROR = "OpenStack access is not configured";
+ private static final Ip4Address DEFAULT_DNS = Ip4Address.valueOf("8.8.8.8");
+ private static final int DHCP_INFINITE_LEASE = -1;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected NetworkConfigRegistry configRegistry;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostProviderRegistry hostProviderRegistry;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DeviceService deviceService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostService hostService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected PacketService packetService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DhcpService dhcpService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected MastershipService mastershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected XosClientService xosClient;
+
+ private final ConfigFactory configFactory =
+ new ConfigFactory(SubjectFactories.APP_SUBJECT_FACTORY, CordVtnConfig.class, "cordvtn") {
+ @Override
+ public CordVtnConfig createConfig() {
+ return new CordVtnConfig();
+ }
+ };
+
+ private final ExecutorService eventExecutor =
+ newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn-instance", "event-handler"));
+ private final PacketProcessor packetProcessor = new InternalPacketProcessor();
+ private final HostListener hostListener = new InternalHostListener();
+ private final NetworkConfigListener configListener = new InternalConfigListener();
+
+ private ApplicationId appId;
+ private HostProviderService hostProvider;
+ private CordVtnArpProxy arpProxy; // TODO make it a component service
+ private MacAddress privateGatewayMac = MacAddress.NONE;
+ private XosAccess xosAccess = null;
+ private OpenStackAccess osAccess = null;
+
+ /**
+ * Creates an cordvtn host location provider.
+ */
+ public CordVtnInstanceManager() {
+ super(new ProviderId("host", CordVtnService.CORDVTN_APP_ID));
+ }
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(CordVtnService.CORDVTN_APP_ID);
+
+ arpProxy = new CordVtnArpProxy(appId, packetService, hostService);
+ packetService.addProcessor(packetProcessor, PacketProcessor.director(0));
+ arpProxy.requestPacket();
+
+ hostService.addListener(hostListener);
+ hostProvider = hostProviderRegistry.register(this);
+
+ configRegistry.registerConfigFactory(configFactory);
+ configRegistry.addListener(configListener);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ hostProviderRegistry.unregister(this);
+ hostService.removeListener(hostListener);
+
+ packetService.removeProcessor(packetProcessor);
+
+ configRegistry.unregisterConfigFactory(configFactory);
+ configRegistry.removeListener(configListener);
+
+ eventExecutor.shutdown();
+ log.info("Stopped");
+ }
+
+ @Override
+ public void triggerProbe(Host host) {
+ /*
+ * Note: In CORD deployment, we assume that all hosts are configured.
+ * Therefore no probe is required.
+ */
+ }
+
+ /**
+ * Adds a service instance at a given connect point.
+ *
+ * @param connectPoint connect point of the instance
+ */
+ public void addInstance(ConnectPoint connectPoint) {
+ Port port = deviceService.getPort(connectPoint.deviceId(), connectPoint.port());
+ if (port == null) {
+ log.debug("No port found from {}", connectPoint);
+ return;
+ }
+
+ VtnPort vtnPort = getVtnPort(port.annotations().value("portName"));
+ if (vtnPort == null) {
+ return;
+ }
+
+ VtnService vtnService = getVtnService(vtnPort.serviceId());
+ if (vtnService == null) {
+ return;
+ }
+
+ // Added CREATE_TIME intentionally to trigger HOST_UPDATED event for the
+ // existing instances.
+ DefaultAnnotations.Builder annotations = DefaultAnnotations.builder()
+ .set(Instance.SERVICE_TYPE, vtnService.serviceType().toString())
+ .set(Instance.SERVICE_ID, vtnPort.serviceId().id())
+ .set(Instance.PORT_ID, vtnPort.id().id())
+ .set(Instance.CREATE_TIME, String.valueOf(System.currentTimeMillis()));
+
+ HostDescription hostDesc = new DefaultHostDescription(
+ vtnPort.mac(),
+ VlanId.NONE,
+ new HostLocation(connectPoint, System.currentTimeMillis()),
+ Sets.newHashSet(vtnPort.ip()),
+ annotations.build());
+
+ HostId hostId = HostId.hostId(vtnPort.mac());
+ hostProvider.hostDetected(hostId, hostDesc, false);
+ }
+
+ /**
+ * Adds a service instance with given host ID and host description.
+ *
+ * @param hostId host id
+ * @param description host description
+ */
+ public void addInstance(HostId hostId, HostDescription description) {
+ hostProvider.hostDetected(hostId, description, false);
+ }
+
+ /**
+ * Removes a service instance from a given connect point.
+ *
+ * @param connectPoint connect point
+ */
+ public void removeInstance(ConnectPoint connectPoint) {
+ hostService.getConnectedHosts(connectPoint)
+ .stream()
+ .forEach(host -> hostProvider.hostVanished(host.id()));
+ }
+
+ /**
+ * Removes service instance with given host ID.
+ *
+ * @param hostId host id
+ */
+ public void removeInstance(HostId hostId) {
+ hostProvider.hostVanished(hostId);
+ }
+
+ private void instanceDetected(Instance instance) {
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ return;
+ }
+
+ if (service.networkType().equals(PRIVATE)) {
+ arpProxy.addGateway(service.serviceIp(), privateGatewayMac);
+ arpProxy.sendGratuitousArpForGateway(service.serviceIp(), Sets.newHashSet(instance));
+ }
+ if (!instance.isNestedInstance()) {
+ registerDhcpLease(instance, service);
+ }
+ }
+
+ private void instanceRemoved(Instance instance) {
+ VtnService service = getVtnService(instance.serviceId());
+ if (service == null) {
+ return;
+ }
+
+ if (service.networkType().equals(PRIVATE) && getInstances(service.id()).isEmpty()) {
+ arpProxy.removeGateway(service.serviceIp());
+ }
+
+ if (!instance.isNestedInstance()) {
+ dhcpService.removeStaticMapping(instance.mac());
+ }
+ }
+
+ private void registerDhcpLease(Instance instance, VtnService service) {
+ Ip4Address broadcast = Ip4Address.makeMaskedAddress(
+ instance.ipAddress(),
+ service.subnet().prefixLength());
+
+ IpAssignment.Builder ipBuilder = IpAssignment.builder()
+ .ipAddress(instance.ipAddress())
+ .leasePeriod(DHCP_INFINITE_LEASE)
+ .timestamp(new Date())
+ .subnetMask(Ip4Address.makeMaskPrefix(service.subnet().prefixLength()))
+ .broadcast(broadcast)
+ .domainServer(DEFAULT_DNS)
+ .assignmentStatus(Option_RangeNotEnforced);
+
+ if (service.networkType() != MANAGEMENT) {
+ ipBuilder = ipBuilder.routerAddress(service.serviceIp().getIp4Address());
+ }
+
+ log.debug("Set static DHCP mapping for {} {}", instance.mac(), instance.ipAddress());
+ dhcpService.setStaticMapping(instance.mac(), ipBuilder.build());
+ }
+
+ private VtnService getVtnService(VtnServiceId serviceId) {
+ checkNotNull(osAccess, OPENSTACK_ACCESS_ERROR);
+ checkNotNull(xosAccess, XOS_ACCESS_ERROR);
+
+ // TODO remove openstack access when XOS provides all information
+ VtnServiceApi serviceApi = xosClient.getClient(xosAccess).vtnService();
+ VtnService service = serviceApi.service(serviceId, osAccess);
+ if (service == null) {
+ log.warn("Failed to get VtnService for {}", serviceId);
+ }
+ return service;
+ }
+
+ private VtnPort getVtnPort(String portName) {
+ checkNotNull(osAccess, OPENSTACK_ACCESS_ERROR);
+ checkNotNull(xosAccess, XOS_ACCESS_ERROR);
+
+ // TODO remove openstack access when XOS provides all information
+ VtnPortApi portApi = xosClient.getClient(xosAccess).vtnPort();
+ VtnPort vtnPort = portApi.vtnPort(portName, osAccess);
+ if (vtnPort == null) {
+ log.warn("Failed to get port information of {}", portName);
+ }
+ return vtnPort;
+ }
+
+ private Set<Instance> getInstances(VtnServiceId serviceId) {
+ return StreamSupport.stream(hostService.getHosts().spliterator(), false)
+ .filter(host -> Objects.equals(
+ serviceId.id(),
+ host.annotations().value(Instance.SERVICE_ID)))
+ .map(Instance::of)
+ .collect(Collectors.toSet());
+ }
+
+ private void readConfiguration() {
+ CordVtnConfig config = configRegistry.getConfig(appId, CordVtnConfig.class);
+ if (config == null) {
+ log.debug("No configuration found");
+ return;
+ }
+
+ log.info("Load CORD-VTN configurations");
+
+ xosAccess = config.xosAccess();
+ osAccess = config.openstackAccess();
+ privateGatewayMac = config.privateGatewayMac();
+
+ Map<IpAddress, MacAddress> publicGateways = config.publicGateways();
+ publicGateways.entrySet()
+ .stream()
+ .forEach(entry -> {
+ arpProxy.addGateway(entry.getKey(), entry.getValue());
+ log.debug("Added public gateway IP {}, MAC {}",
+ entry.getKey(), entry.getValue());
+ });
+ // TODO notice gateway MAC change to VMs holds this gateway IP
+ }
+
+ private class InternalHostListener implements HostListener {
+
+ @Override
+ public void event(HostEvent event) {
+ Host host = event.subject();
+ if (!mastershipService.isLocalMaster(host.location().deviceId())) {
+ // do not allow to proceed without mastership
+ return;
+ }
+
+ Instance instance = Instance.of(host);
+ switch (event.type()) {
+ case HOST_UPDATED:
+ case HOST_ADDED:
+ eventExecutor.execute(() -> instanceDetected(instance));
+ break;
+ case HOST_REMOVED:
+ eventExecutor.execute(() -> instanceRemoved(instance));
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ private class InternalPacketProcessor implements PacketProcessor {
+
+ @Override
+ public void process(PacketContext context) {
+ if (context.isHandled()) {
+ return;
+ }
+ Ethernet ethPacket = context.inPacket().parsed();
+ if (ethPacket == null || ethPacket.getEtherType() != Ethernet.TYPE_ARP) {
+ return;
+ }
+ arpProxy.processArpPacket(context, ethPacket);
+ }
+ }
+
+ private class InternalConfigListener implements NetworkConfigListener {
+
+ @Override
+ public void event(NetworkConfigEvent event) {
+ if (!event.configClass().equals(CordVtnConfig.class)) {
+ return;
+ }
+
+ switch (event.type()) {
+ case CONFIG_ADDED:
+ case CONFIG_UPDATED:
+ readConfiguration();
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtnNodeManager.java b/src/main/java/org/opencord/cordvtn/impl/CordVtnNodeManager.java
new file mode 100644
index 0000000..c037c05
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtnNodeManager.java
@@ -0,0 +1,1029 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import com.google.common.collect.Sets;
+import com.jcraft.jsch.Session;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.IpAddress;
+import org.onlab.util.ItemNotFoundException;
+import org.onlab.util.KryoNamespace;
+import org.onosproject.cluster.ClusterService;
+import org.onosproject.cluster.LeadershipService;
+import org.onosproject.cluster.NodeId;
+import org.opencord.cordvtn.api.ConnectionHandler;
+import org.opencord.cordvtn.api.CordVtnConfig;
+import org.opencord.cordvtn.api.CordVtnNode;
+import org.opencord.cordvtn.api.CordVtnNodeState;
+import org.opencord.cordvtn.api.CordVtnService;
+import org.opencord.cordvtn.api.NetworkAddress;
+import org.opencord.cordvtn.api.SshAccessInfo;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.net.ConnectPoint;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.Host;
+import org.onosproject.net.Port;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.behaviour.BridgeConfig;
+import org.onosproject.net.behaviour.BridgeName;
+import org.onosproject.net.behaviour.ControllerInfo;
+import org.onosproject.net.behaviour.DefaultTunnelDescription;
+import org.onosproject.net.behaviour.TunnelConfig;
+import org.onosproject.net.behaviour.TunnelDescription;
+import org.onosproject.net.behaviour.TunnelName;
+import org.onosproject.net.config.NetworkConfigEvent;
+import org.onosproject.net.config.NetworkConfigListener;
+import org.onosproject.net.config.NetworkConfigRegistry;
+import org.onosproject.net.config.NetworkConfigService;
+import org.onosproject.net.device.DeviceAdminService;
+import org.onosproject.net.device.DeviceEvent;
+import org.onosproject.net.device.DeviceListener;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.host.HostService;
+import org.onosproject.ovsdb.controller.OvsdbClientService;
+import org.onosproject.ovsdb.controller.OvsdbController;
+import org.onosproject.ovsdb.controller.OvsdbNodeId;
+import org.onosproject.store.serializers.KryoNamespaces;
+import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
+import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
+import org.slf4j.Logger;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.opencord.cordvtn.impl.CordVtnPipeline.DEFAULT_TUNNEL;
+import static org.onosproject.net.Device.Type.SWITCH;
+import static org.onosproject.net.behaviour.TunnelDescription.Type.VXLAN;
+import static org.opencord.cordvtn.impl.RemoteIpCommandUtil.*;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Reads node information from the network config file and handles the config
+ * update events.
+ * Only a leader controller performs the node addition or deletion.
+ */
+@Component(immediate = true)
+@Service(value = CordVtnNodeManager.class)
+public class CordVtnNodeManager {
+
+ protected final Logger log = getLogger(getClass());
+
+ private static final KryoNamespace.Builder NODE_SERIALIZER = KryoNamespace.newBuilder()
+ .register(KryoNamespaces.API)
+ .register(CordVtnNode.class)
+ .register(NodeState.class)
+ .register(SshAccessInfo.class)
+ .register(NetworkAddress.class);
+
+ private static final String DEFAULT_BRIDGE = "br-int";
+ private static final String VPORT_PREFIX = "tap";
+ private static final String OK = "OK";
+ private static final String NO = "NO";
+
+ private static final Map<String, String> DEFAULT_TUNNEL_OPTIONS = new HashMap<String, String>() {
+ {
+ put("key", "flow");
+ put("remote_ip", "flow");
+ }
+ };
+ private static final int DPID_BEGIN = 3;
+ private static final int OFPORT = 6653;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected NetworkConfigRegistry configRegistry;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected NetworkConfigService configService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected StorageService storageService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DeviceAdminService adminService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected OvsdbController controller;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected ClusterService clusterService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DeviceService deviceService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected HostService hostService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected LeadershipService leadershipService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CordVtnInstanceManager instanceManager;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CordVtnPipeline pipeline;
+
+ private final ExecutorService eventExecutor =
+ newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn-node", "event-handler"));
+
+ private final NetworkConfigListener configListener = new InternalConfigListener();
+ private final DeviceListener deviceListener = new InternalDeviceListener();
+ private final MapEventListener<String, CordVtnNode> nodeStoreListener = new InternalMapListener();
+
+ private final OvsdbHandler ovsdbHandler = new OvsdbHandler();
+ private final BridgeHandler bridgeHandler = new BridgeHandler();
+
+ private ConsistentMap<String, CordVtnNode> nodeStore;
+ private ApplicationId appId;
+ private NodeId localNodeId;
+
+ private enum NodeState implements CordVtnNodeState {
+
+ INIT {
+ @Override
+ public void process(CordVtnNodeManager nodeManager, CordVtnNode node) {
+ if (!nodeManager.isOvsdbConnected(node)) {
+ nodeManager.connectOvsdb(node);
+ } else {
+ nodeManager.createIntegrationBridge(node);
+ }
+ }
+ },
+ BRIDGE_CREATED {
+ @Override
+ public void process(CordVtnNodeManager nodeManager, CordVtnNode node) {
+ if (!nodeManager.isOvsdbConnected(node)) {
+ nodeManager.connectOvsdb(node);
+ } else {
+ nodeManager.createTunnelInterface(node);
+ nodeManager.addDataPlaneInterface(node);
+ }
+ }
+ },
+ PORTS_ADDED {
+ @Override
+ public void process(CordVtnNodeManager nodeManager, CordVtnNode node) {
+ nodeManager.setIpAddress(node);
+ }
+ },
+ COMPLETE {
+ @Override
+ public void process(CordVtnNodeManager nodeManager, CordVtnNode node) {
+ nodeManager.postInit(node);
+ }
+ },
+ INCOMPLETE {
+ @Override
+ public void process(CordVtnNodeManager nodeManager, CordVtnNode node) {
+ }
+ };
+
+ public abstract void process(CordVtnNodeManager nodeManager, CordVtnNode node);
+ }
+
+ @Activate
+ protected void activate() {
+ appId = coreService.getAppId(CordVtnService.CORDVTN_APP_ID);
+
+ localNodeId = clusterService.getLocalNode().id();
+ leadershipService.runForLeadership(appId.name());
+
+ nodeStore = storageService.<String, CordVtnNode>consistentMapBuilder()
+ .withSerializer(Serializer.using(NODE_SERIALIZER.build()))
+ .withName("cordvtn-nodestore")
+ .withApplicationId(appId)
+ .build();
+
+ nodeStore.addListener(nodeStoreListener);
+ deviceService.addListener(deviceListener);
+ configService.addListener(configListener);
+
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ configService.removeListener(configListener);
+ deviceService.removeListener(deviceListener);
+ nodeStore.removeListener(nodeStoreListener);
+
+ leadershipService.withdraw(appId.name());
+ eventExecutor.shutdown();
+
+ log.info("Stopped");
+ }
+
+ /**
+ * Adds or updates a new node to the service.
+ *
+ * @param node cordvtn node
+ */
+ public void addOrUpdateNode(CordVtnNode node) {
+ checkNotNull(node);
+ nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, getNodeState(node)));
+ }
+
+ /**
+ * Deletes a node from the service.
+ *
+ * @param node cordvtn node
+ */
+ public void deleteNode(CordVtnNode node) {
+ checkNotNull(node);
+
+ if (isOvsdbConnected(node)) {
+ disconnectOvsdb(node);
+ }
+
+ nodeStore.remove(node.hostname());
+ }
+
+ /**
+ * Returns node initialization state.
+ *
+ * @param node cordvtn node
+ * @return true if initial node setup is completed, otherwise false
+ */
+ public boolean isNodeInitComplete(CordVtnNode node) {
+ checkNotNull(node);
+ return nodeStore.containsKey(node.hostname()) && getNodeState(node).equals(NodeState.COMPLETE);
+ }
+
+ /**
+ * Returns detailed node initialization state.
+ *
+ * @param node cordvtn node
+ * @return string including detailed node init state
+ */
+ public String checkNodeInitState(CordVtnNode node) {
+ checkNotNull(node);
+
+ if (!nodeStore.containsKey(node.hostname())) {
+ log.warn("Node {} does not exist, add node first", node.hostname());
+ return null;
+ }
+
+ Session session = connect(node.sshInfo());
+ if (session == null) {
+ log.debug("Failed to SSH to {}", node.hostname());
+ return null;
+ }
+
+ Set<IpAddress> intBrIps = getCurrentIps(session, DEFAULT_BRIDGE);
+ String result = String.format(
+ "Current state : %s%n" +
+ "br-int created and connected to ONOS : %s (%s)%n" +
+ "VXLAN interface added to br-int : %s%n" +
+ "Data plane interface is added to br-int and enabled : %s (%s)%n" +
+ "IP flushed from data plane interface : %s (%s)%n" +
+ "Data plane IP added to br-int : %s (%s)%n" +
+ "Local management IP added to br-int : %s (%s)",
+ node.state(),
+ isBrIntCreated(node) ? OK : NO, node.intBrId(),
+ isTunnelIntfCreated(node) ? OK : NO,
+ isDataPlaneIntfAdded(node) ? OK : NO, node.dpIntf(),
+ isInterfaceUp(session, node.dpIntf()) &&
+ getCurrentIps(session, node.dpIntf()).isEmpty() ? OK : NO, node.dpIntf(),
+ intBrIps.contains(node.dpIp().ip()) ? OK : NO, node.dpIp().cidr(),
+ intBrIps.contains(node.localMgmtIp().ip()) ? OK : NO, node.localMgmtIp().cidr());
+
+ disconnect(session);
+
+ return result;
+ }
+
+ /**
+ * Returns the number of the nodes known to the service.
+ *
+ * @return number of nodes
+ */
+ public int getNodeCount() {
+ return nodeStore.size();
+ }
+
+ /**
+ * Returns all nodes known to the service.
+ *
+ * @return list of nodes
+ */
+ public List<CordVtnNode> getNodes() {
+ return nodeStore.values().stream().map(Versioned::value).collect(Collectors.toList());
+ }
+
+ /**
+ * Returns all nodes in complete state.
+ *
+ * @return set of nodes
+ */
+ public Set<CordVtnNode> completeNodes() {
+ return getNodes().stream().filter(this::isNodeInitComplete).collect(Collectors.toSet());
+ }
+
+ /**
+ * Returns physical data plane port number of a given device.
+ *
+ * @param deviceId integration bridge device id
+ * @return port number; null otherwise
+ */
+ public PortNumber dpPort(DeviceId deviceId) {
+ CordVtnNode node = nodeByBridgeId(deviceId);
+ if (node == null) {
+ log.warn("Failed to get node for {}", deviceId);
+ return null;
+ }
+ Port port = deviceService.getPorts(deviceId).stream()
+ .filter(p -> portName(p).contains(node.dpIntf()) &&
+ p.isEnabled())
+ .findFirst().orElse(null);
+
+ return port == null ? null : port.number();
+ }
+
+ /**
+ * Returns physical data plane IP address of a given device.
+ *
+ * @param deviceId integration bridge device id
+ * @return ip address; null otherwise
+ */
+ public IpAddress dpIp(DeviceId deviceId) {
+ CordVtnNode node = nodeByBridgeId(deviceId);
+ if (node == null) {
+ log.warn("Failed to get node for {}", deviceId);
+ return null;
+ }
+ return node.dpIp().ip();
+ }
+
+ /**
+ * Returns tunnel port number of a given device.
+ *
+ * @param deviceId integration bridge device id
+ * @return port number
+ */
+ public PortNumber tunnelPort(DeviceId deviceId) {
+ Port port = deviceService.getPorts(deviceId).stream()
+ .filter(p -> portName(p).contains(DEFAULT_TUNNEL))
+ .findFirst().orElse(null);
+
+ return port == null ? null : port.number();
+ }
+
+ /**
+ * Returns if current node state saved in nodeStore is COMPLETE or not.
+ *
+ * @param node cordvtn node
+ * @return true if it's complete state, otherwise false
+ */
+ private boolean isNodeStateComplete(CordVtnNode node) {
+ checkNotNull(node);
+
+ // the state saved in nodeStore can be wrong if IP address settings are changed
+ // after the node init has been completed since there's no way to detect it
+ // getNodeState and checkNodeInitState always return correct answer but can be slow
+ Versioned<CordVtnNode> versionedNode = nodeStore.get(node.hostname());
+ CordVtnNodeState state = versionedNode.value().state();
+ return state != null && state.equals(NodeState.COMPLETE);
+ }
+
+ /**
+ * Initiates node to serve virtual tenant network.
+ *
+ * @param node cordvtn node
+ */
+ private void initNode(CordVtnNode node) {
+ checkNotNull(node);
+
+ NodeState state = (NodeState) node.state();
+ log.debug("Processing node: {} state: {}", node.hostname(), state);
+
+ state.process(this, node);
+ }
+
+ /**
+ * Performs tasks after node initialization.
+ * It disconnects unnecessary OVSDB connection and installs initial flow
+ * rules on the device.
+ *
+ * @param node cordvtn node
+ */
+ private void postInit(CordVtnNode node) {
+ disconnectOvsdb(node);
+ pipeline.initPipeline(node, dpPort(node.intBrId()), tunnelPort(node.intBrId()));
+
+ deviceService.getPorts(node.intBrId()).stream()
+ .filter(port -> portName(port).startsWith(VPORT_PREFIX) &&
+ port.isEnabled())
+ .forEach(port -> instanceManager.addInstance(connectPoint(port)));
+
+ hostService.getHosts().forEach(host -> {
+ if (deviceService.getPort(host.location().deviceId(),
+ host.location().port()) == null) {
+ instanceManager.removeInstance(connectPoint(host));
+ }
+ });
+
+ log.info("Finished init {}", node.hostname());
+ }
+
+ /**
+ * Sets a new state for a given cordvtn node.
+ *
+ * @param node cordvtn node
+ * @param newState new node state
+ */
+ private void setNodeState(CordVtnNode node, NodeState newState) {
+ checkNotNull(node);
+
+ log.debug("Changed {} state: {}", node.hostname(), newState);
+ nodeStore.put(node.hostname(), CordVtnNode.getUpdatedNode(node, newState));
+ }
+
+ /**
+ * Checks current state of a given cordvtn node and returns it.
+ *
+ * @param node cordvtn node
+ * @return node state
+ */
+ private NodeState getNodeState(CordVtnNode node) {
+ checkNotNull(node);
+
+ if (isBrIntCreated(node) && isTunnelIntfCreated(node) &&
+ isDataPlaneIntfAdded(node) && isIpAddressSet(node)) {
+ return NodeState.COMPLETE;
+ } else if (isDataPlaneIntfAdded(node) && isTunnelIntfCreated(node)) {
+ return NodeState.PORTS_ADDED;
+ } else if (isBrIntCreated(node)) {
+ return NodeState.BRIDGE_CREATED;
+ } else {
+ return NodeState.INIT;
+ }
+ }
+
+ /**
+ * Returns connection state of OVSDB server for a given node.
+ *
+ * @param node cordvtn node
+ * @return true if it is connected, false otherwise
+ */
+ private boolean isOvsdbConnected(CordVtnNode node) {
+ checkNotNull(node);
+
+ OvsdbClientService ovsdbClient = getOvsdbClient(node);
+ return deviceService.isAvailable(node.ovsdbId()) &&
+ ovsdbClient != null && ovsdbClient.isConnected();
+ }
+
+ /**
+ * Connects to OVSDB server for a given node.
+ *
+ * @param node cordvtn node
+ */
+ private void connectOvsdb(CordVtnNode node) {
+ checkNotNull(node);
+
+ if (!nodeStore.containsKey(node.hostname())) {
+ log.warn("Node {} does not exist", node.hostname());
+ return;
+ }
+
+ if (!isOvsdbConnected(node)) {
+ controller.connect(node.hostMgmtIp().ip(), node.ovsdbPort());
+ }
+ }
+
+ /**
+ * Disconnects OVSDB server for a given node.
+ *
+ * @param node cordvtn node
+ */
+ private void disconnectOvsdb(CordVtnNode node) {
+ checkNotNull(node);
+
+ if (!nodeStore.containsKey(node.hostname())) {
+ log.warn("Node {} does not exist", node.hostname());
+ return;
+ }
+
+ if (isOvsdbConnected(node)) {
+ OvsdbClientService ovsdbClient = getOvsdbClient(node);
+ ovsdbClient.disconnect();
+ }
+ }
+
+ /**
+ * Returns OVSDB client for a given node.
+ *
+ * @param node cordvtn node
+ * @return OVSDB client, or null if it fails to get OVSDB client
+ */
+ private OvsdbClientService getOvsdbClient(CordVtnNode node) {
+ checkNotNull(node);
+
+ OvsdbClientService ovsdbClient = controller.getOvsdbClient(
+ new OvsdbNodeId(node.hostMgmtIp().ip(), node.ovsdbPort().toInt()));
+ if (ovsdbClient == null) {
+ log.trace("Couldn't find OVSDB client for {}", node.hostname());
+ }
+ return ovsdbClient;
+ }
+
+ /**
+ * Creates an integration bridge for a given node.
+ *
+ * @param node cordvtn node
+ */
+ private void createIntegrationBridge(CordVtnNode node) {
+ if (isBrIntCreated(node)) {
+ return;
+ }
+
+ List<ControllerInfo> controllers = new ArrayList<>();
+ Sets.newHashSet(clusterService.getNodes()).stream()
+ .forEach(controller -> {
+ ControllerInfo ctrlInfo = new ControllerInfo(controller.ip(), OFPORT, "tcp");
+ controllers.add(ctrlInfo);
+ });
+
+ String dpid = node.intBrId().toString().substring(DPID_BEGIN);
+
+ try {
+ Device device = deviceService.getDevice(node.ovsdbId());
+ if (device.is(BridgeConfig.class)) {
+ BridgeConfig bridgeConfig = device.as(BridgeConfig.class);
+ bridgeConfig.addBridge(BridgeName.bridgeName(DEFAULT_BRIDGE), dpid, controllers);
+ } else {
+ log.warn("The bridging behaviour is not supported in device {}", device.id());
+ }
+ } catch (ItemNotFoundException e) {
+ log.warn("Failed to create integration bridge on {}", node.hostname());
+ }
+ }
+
+ /**
+ * Creates tunnel interface to the integration bridge for a given node.
+ *
+ * @param node cordvtn node
+ */
+ private void createTunnelInterface(CordVtnNode node) {
+ if (isTunnelIntfCreated(node)) {
+ return;
+ }
+
+ DefaultAnnotations.Builder optionBuilder = DefaultAnnotations.builder();
+ for (String key : DEFAULT_TUNNEL_OPTIONS.keySet()) {
+ optionBuilder.set(key, DEFAULT_TUNNEL_OPTIONS.get(key));
+ }
+
+ TunnelDescription description = new DefaultTunnelDescription(
+ null, null, VXLAN, TunnelName.tunnelName(DEFAULT_TUNNEL),
+ optionBuilder.build());
+
+ try {
+ Device device = deviceService.getDevice(node.ovsdbId());
+ if (device.is(TunnelConfig.class)) {
+ TunnelConfig tunnelConfig = device.as(TunnelConfig.class);
+ tunnelConfig.createTunnelInterface(BridgeName.bridgeName(DEFAULT_BRIDGE), description);
+ } else {
+ log.warn("The tunneling behaviour is not supported in device {}", device.id());
+ }
+ } catch (ItemNotFoundException e) {
+ log.warn("Failed to create tunnel interface on {}", node.hostname());
+ }
+ }
+
+ /**
+ * Adds data plane interface to a given node.
+ *
+ * @param node cordvtn node
+ */
+ private void addDataPlaneInterface(CordVtnNode node) {
+ if (isDataPlaneIntfAdded(node)) {
+ return;
+ }
+
+ Session session = connect(node.sshInfo());
+ if (session == null) {
+ log.debug("Failed to SSH to {}", node.hostname());
+ return;
+ }
+
+ if (!isInterfaceUp(session, node.dpIntf())) {
+ log.warn("Interface {} is not available", node.dpIntf());
+ return;
+ }
+ disconnect(session);
+
+ try {
+ Device device = deviceService.getDevice(node.ovsdbId());
+ if (device.is(BridgeConfig.class)) {
+ BridgeConfig bridgeConfig = device.as(BridgeConfig.class);
+ bridgeConfig.addPort(BridgeName.bridgeName(DEFAULT_BRIDGE), node.dpIntf());
+ } else {
+ log.warn("The bridging behaviour is not supported in device {}", device.id());
+ }
+ } catch (ItemNotFoundException e) {
+ log.warn("Failed to add {} on {}", node.dpIntf(), node.hostname());
+ }
+ }
+
+ /**
+ * Flushes IP address from data plane interface and adds data plane IP address
+ * to integration bridge.
+ *
+ * @param node cordvtn node
+ */
+ private void setIpAddress(CordVtnNode node) {
+ Session session = connect(node.sshInfo());
+ if (session == null) {
+ log.debug("Failed to SSH to {}", node.hostname());
+ return;
+ }
+
+ getCurrentIps(session, DEFAULT_BRIDGE).stream()
+ .filter(ip -> !ip.equals(node.localMgmtIp().ip()))
+ .filter(ip -> !ip.equals(node.dpIp().ip()))
+ .forEach(ip -> deleteIp(session, ip, DEFAULT_BRIDGE));
+
+ boolean result = flushIp(session, node.dpIntf()) &&
+ setInterfaceUp(session, node.dpIntf()) &&
+ addIp(session, node.dpIp(), DEFAULT_BRIDGE) &&
+ addIp(session, node.localMgmtIp(), DEFAULT_BRIDGE) &&
+ setInterfaceUp(session, DEFAULT_BRIDGE);
+
+ disconnect(session);
+
+ if (result) {
+ setNodeState(node, NodeState.COMPLETE);
+ }
+ }
+
+ /**
+ * Checks if integration bridge exists and available.
+ *
+ * @param node cordvtn node
+ * @return true if the bridge is available, false otherwise
+ */
+ private boolean isBrIntCreated(CordVtnNode node) {
+ return (deviceService.getDevice(node.intBrId()) != null
+ && deviceService.isAvailable(node.intBrId()));
+ }
+
+ /**
+ * Checks if tunnel interface exists.
+ *
+ * @param node cordvtn node
+ * @return true if the interface exists, false otherwise
+ */
+ private boolean isTunnelIntfCreated(CordVtnNode node) {
+ return deviceService.getPorts(node.intBrId())
+ .stream()
+ .filter(p -> portName(p).contains(DEFAULT_TUNNEL) &&
+ p.isEnabled())
+ .findAny().isPresent();
+ }
+
+ /**
+ * Checks if data plane interface exists.
+ *
+ * @param node cordvtn node
+ * @return true if the interface exists, false otherwise
+ */
+ private boolean isDataPlaneIntfAdded(CordVtnNode node) {
+ return deviceService.getPorts(node.intBrId())
+ .stream()
+ .filter(p -> portName(p).contains(node.dpIntf()) &&
+ p.isEnabled())
+ .findAny().isPresent();
+ }
+
+ /**
+ * Checks if the IP addresses are correctly set.
+ *
+ * @param node cordvtn node
+ * @return true if the IP is set, false otherwise
+ */
+ private boolean isIpAddressSet(CordVtnNode node) {
+ Session session = connect(node.sshInfo());
+ if (session == null) {
+ log.debug("Failed to SSH to {}", node.hostname());
+ return false;
+ }
+
+ Set<IpAddress> intBrIps = getCurrentIps(session, DEFAULT_BRIDGE);
+ boolean result = getCurrentIps(session, node.dpIntf()).isEmpty() &&
+ isInterfaceUp(session, node.dpIntf()) &&
+ intBrIps.contains(node.dpIp().ip()) &&
+ intBrIps.contains(node.localMgmtIp().ip()) &&
+ isInterfaceUp(session, DEFAULT_BRIDGE);
+
+ disconnect(session);
+ return result;
+ }
+
+ /**
+ * Returns connect point of a given port.
+ *
+ * @param port port
+ * @return connect point
+ */
+ private ConnectPoint connectPoint(Port port) {
+ return new ConnectPoint(port.element().id(), port.number());
+ }
+
+ /**
+ * Returns connect point of a given host.
+ *
+ * @param host host
+ * @return connect point
+ */
+ private ConnectPoint connectPoint(Host host) {
+ return new ConnectPoint(host.location().deviceId(), host.location().port());
+ }
+
+ /**
+ * Returns cordvtn node associated with a given OVSDB device.
+ *
+ * @param ovsdbId OVSDB device id
+ * @return cordvtn node, null if it fails to find the node
+ */
+ private CordVtnNode nodeByOvsdbId(DeviceId ovsdbId) {
+ return getNodes().stream()
+ .filter(node -> node.ovsdbId().equals(ovsdbId))
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * Returns cordvtn node associated with a given integration bridge.
+ *
+ * @param bridgeId device id of integration bridge
+ * @return cordvtn node, null if it fails to find the node
+ */
+ private CordVtnNode nodeByBridgeId(DeviceId bridgeId) {
+ return getNodes().stream()
+ .filter(node -> node.intBrId().equals(bridgeId))
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * Returns port name.
+ *
+ * @param port port
+ * @return port name
+ */
+ private String portName(Port port) {
+ return port.annotations().value("portName");
+ }
+
+ private class OvsdbHandler implements ConnectionHandler<Device> {
+
+ @Override
+ public void connected(Device device) {
+ CordVtnNode node = nodeByOvsdbId(device.id());
+ if (node != null) {
+ setNodeState(node, getNodeState(node));
+ } else {
+ log.debug("{} is detected on unregistered node, ignore it.", device.id());
+ }
+ }
+
+ @Override
+ public void disconnected(Device device) {
+ if (!deviceService.isAvailable(device.id())) {
+ log.debug("Device {} is disconnected", device.id());
+ adminService.removeDevice(device.id());
+ }
+ }
+ }
+
+ private class BridgeHandler implements ConnectionHandler<Device> {
+
+ @Override
+ public void connected(Device device) {
+ CordVtnNode node = nodeByBridgeId(device.id());
+ if (node != null) {
+ setNodeState(node, getNodeState(node));
+ } else {
+ log.debug("{} is detected on unregistered node, ignore it.", device.id());
+ }
+ }
+
+ @Override
+ public void disconnected(Device device) {
+ CordVtnNode node = nodeByBridgeId(device.id());
+ if (node != null) {
+ log.debug("Integration Bridge is disconnected from {}", node.hostname());
+ setNodeState(node, NodeState.INCOMPLETE);
+ }
+ }
+
+ /**
+ * Handles port added situation.
+ * If the added port is tunnel or data plane interface, proceed to the remaining
+ * node initialization. Otherwise, do nothing.
+ *
+ * @param port port
+ */
+ public void portAdded(Port port) {
+ CordVtnNode node = nodeByBridgeId((DeviceId) port.element().id());
+ String portName = portName(port);
+
+ if (node == null) {
+ log.debug("{} is added to unregistered node, ignore it.", portName);
+ return;
+ }
+
+ log.info("Port {} is added to {}", portName, node.hostname());
+
+ if (portName.startsWith(VPORT_PREFIX)) {
+ if (isNodeStateComplete(node)) {
+ instanceManager.addInstance(connectPoint(port));
+ } else {
+ log.debug("VM is detected on incomplete node, ignore it.", portName);
+ }
+ } else if (portName.contains(DEFAULT_TUNNEL) || portName.equals(node.dpIntf())) {
+ setNodeState(node, getNodeState(node));
+ }
+ }
+
+ /**
+ * Handles port removed situation.
+ * If the removed port is tunnel or data plane interface, proceed to the remaining
+ * node initialization.Others, do nothing.
+ *
+ * @param port port
+ */
+ public void portRemoved(Port port) {
+ CordVtnNode node = nodeByBridgeId((DeviceId) port.element().id());
+ String portName = portName(port);
+
+ if (node == null) {
+ return;
+ }
+
+ log.info("Port {} is removed from {}", portName, node.hostname());
+
+ if (portName.startsWith(VPORT_PREFIX)) {
+ if (isNodeStateComplete(node)) {
+ instanceManager.removeInstance(connectPoint(port));
+ } else {
+ log.debug("VM is vanished from incomplete node, ignore it.", portName);
+ }
+ } else if (portName.contains(DEFAULT_TUNNEL) || portName.equals(node.dpIntf())) {
+ setNodeState(node, NodeState.INCOMPLETE);
+ }
+ }
+ }
+
+ private class InternalDeviceListener implements DeviceListener {
+
+ @Override
+ public void event(DeviceEvent event) {
+
+ NodeId leaderNodeId = leadershipService.getLeader(appId.name());
+ if (!Objects.equals(localNodeId, leaderNodeId)) {
+ // do not allow to proceed without leadership
+ return;
+ }
+
+ Device device = event.subject();
+ ConnectionHandler<Device> handler =
+ (device.type().equals(SWITCH) ? bridgeHandler : ovsdbHandler);
+
+ switch (event.type()) {
+ case PORT_ADDED:
+ eventExecutor.execute(() -> bridgeHandler.portAdded(event.port()));
+ break;
+ case PORT_UPDATED:
+ if (!event.port().isEnabled()) {
+ eventExecutor.execute(() -> bridgeHandler.portRemoved(event.port()));
+ }
+ break;
+ case DEVICE_ADDED:
+ case DEVICE_AVAILABILITY_CHANGED:
+ if (deviceService.isAvailable(device.id())) {
+ eventExecutor.execute(() -> handler.connected(device));
+ } else {
+ eventExecutor.execute(() -> handler.disconnected(device));
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ /**
+ * Reads cordvtn nodes from config file.
+ */
+ private void readConfiguration() {
+ CordVtnConfig config = configRegistry.getConfig(appId, CordVtnConfig.class);
+ if (config == null) {
+ log.debug("No configuration found");
+ return;
+ }
+ config.cordVtnNodes().forEach(this::addOrUpdateNode);
+ }
+
+ private class InternalConfigListener implements NetworkConfigListener {
+
+ @Override
+ public void event(NetworkConfigEvent event) {
+ NodeId leaderNodeId = leadershipService.getLeader(appId.name());
+ if (!Objects.equals(localNodeId, leaderNodeId)) {
+ // do not allow to proceed without leadership
+ return;
+ }
+
+ if (!event.configClass().equals(CordVtnConfig.class)) {
+ return;
+ }
+
+ switch (event.type()) {
+ case CONFIG_ADDED:
+ case CONFIG_UPDATED:
+ eventExecutor.execute(CordVtnNodeManager.this::readConfiguration);
+ break;
+ default:
+ break;
+ }
+ }
+ }
+
+ private class InternalMapListener implements MapEventListener<String, CordVtnNode> {
+
+ @Override
+ public void event(MapEvent<String, CordVtnNode> event) {
+ NodeId leaderNodeId = leadershipService.getLeader(appId.name());
+ if (!Objects.equals(localNodeId, leaderNodeId)) {
+ // do not allow to proceed without leadership
+ return;
+ }
+
+ CordVtnNode oldNode;
+ CordVtnNode newNode;
+
+ switch (event.type()) {
+ case UPDATE:
+ oldNode = event.oldValue().value();
+ newNode = event.newValue().value();
+
+ log.info("Reloaded {}", newNode.hostname());
+ if (!newNode.equals(oldNode)) {
+ log.debug("New node: {}", newNode);
+ }
+ // performs init procedure even if the node is not changed
+ // for robustness since it's no harm to run init procedure
+ // multiple times
+ eventExecutor.execute(() -> initNode(newNode));
+ break;
+ case INSERT:
+ newNode = event.newValue().value();
+ log.info("Added {}", newNode.hostname());
+ eventExecutor.execute(() -> initNode(newNode));
+ break;
+ case REMOVE:
+ oldNode = event.oldValue().value();
+ log.info("Removed {}", oldNode.hostname());
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/CordVtnPipeline.java b/src/main/java/org/opencord/cordvtn/impl/CordVtnPipeline.java
new file mode 100644
index 0000000..f5799c3
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/CordVtnPipeline.java
@@ -0,0 +1,409 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IPv4;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.TpPort;
+import org.onlab.packet.VlanId;
+import org.onlab.util.ItemNotFoundException;
+import org.opencord.cordvtn.api.CordVtnNode;
+import org.opencord.cordvtn.api.CordVtnService;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.CoreService;
+import org.onosproject.net.Device;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.behaviour.ExtensionTreatmentResolver;
+import org.onosproject.net.device.DeviceService;
+import org.onosproject.net.flow.DefaultFlowRule;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.FlowRuleOperations;
+import org.onosproject.net.flow.FlowRuleOperationsContext;
+import org.onosproject.net.flow.FlowRuleService;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.flow.instructions.ExtensionPropertyException;
+import org.onosproject.net.flow.instructions.ExtensionTreatment;
+import org.slf4j.Logger;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onosproject.net.flow.instructions.ExtensionTreatmentType.ExtensionTreatmentTypes.NICIRA_SET_TUNNEL_DST;
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * Provides CORD VTN pipeline.
+ */
+@Component(immediate = true)
+@Service(value = CordVtnPipeline.class)
+public final class CordVtnPipeline {
+
+ protected final Logger log = getLogger(getClass());
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CoreService coreService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected FlowRuleService flowRuleService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected DeviceService deviceService;
+
+ // tables
+ public static final int TABLE_ZERO = 0;
+ public static final int TABLE_IN_PORT = 1;
+ public static final int TABLE_ACCESS_TYPE = 2;
+ public static final int TABLE_IN_SERVICE = 3;
+ public static final int TABLE_DST_IP = 4;
+ public static final int TABLE_TUNNEL_IN = 5;
+ public static final int TABLE_VLAN = 6;
+
+ // priorities
+ public static final int PRIORITY_MANAGEMENT = 55000;
+ public static final int PRIORITY_HIGH = 50000;
+ public static final int PRIORITY_DEFAULT = 5000;
+ public static final int PRIORITY_LOW = 4000;
+ public static final int PRIORITY_ZERO = 0;
+
+ public static final int VXLAN_UDP_PORT = 4789;
+ public static final VlanId VLAN_WAN = VlanId.vlanId((short) 500);
+ public static final String DEFAULT_TUNNEL = "vxlan";
+ private static final String PORT_NAME = "portName";
+
+ private ApplicationId appId;
+
+ @Activate
+ protected void activate() {
+ appId = coreService.registerApplication(CordVtnService.CORDVTN_APP_ID);
+ log.info("Started");
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ log.info("Stopped");
+ }
+
+ /**
+ * Flush flows installed by this application.
+ */
+ public void flushRules() {
+ flowRuleService.getFlowRulesById(appId).forEach(flowRule -> processFlowRule(false, flowRule));
+ }
+
+ /**
+ * Installs table miss rule to a give device.
+ *
+ * @param node cordvtn node
+ * @param dpPort data plane port number
+ * @param tunnelPort tunnel port number
+ */
+ public void initPipeline(CordVtnNode node, PortNumber dpPort, PortNumber tunnelPort) {
+ checkNotNull(node);
+
+ processTableZero(node.intBrId(), dpPort, node.dpIp().ip());
+ processInPortTable(node.intBrId(), tunnelPort, dpPort);
+ processAccessTypeTable(node.intBrId(), dpPort);
+ processVlanTable(node.intBrId(), dpPort);
+ }
+
+ private void processTableZero(DeviceId deviceId, PortNumber dpPort, IpAddress dpIp) {
+ // take vxlan packet out onto the physical port
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(PortNumber.LOCAL)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(dpPort)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ // take a vxlan encap'd packet through the Linux stack
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(dpPort)
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPProtocol(IPv4.PROTOCOL_UDP)
+ .matchUdpDst(TpPort.tpPort(VXLAN_UDP_PORT))
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ // take a packet to the data plane ip through Linux stack
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(dpPort)
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(dpIp.toIpPrefix())
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ // take an arp packet from physical through Linux stack
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(dpPort)
+ .matchEthType(Ethernet.TYPE_ARP)
+ .matchArpTpa(dpIp.getIp4Address())
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ // take all else to the next table
+ selector = DefaultTrafficSelector.builder()
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .transition(TABLE_IN_PORT)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_ZERO)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ // take all vlan tagged packet to the VLAN table
+ selector = DefaultTrafficSelector.builder()
+ .matchVlanId(VlanId.ANY)
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .transition(TABLE_VLAN)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_MANAGEMENT)
+ .forDevice(deviceId)
+ .forTable(TABLE_ZERO)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+ }
+
+ private void processInPortTable(DeviceId deviceId, PortNumber tunnelPort, PortNumber dpPort) {
+ checkNotNull(tunnelPort);
+
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(tunnelPort)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .transition(TABLE_TUNNEL_IN)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_DEFAULT)
+ .forDevice(deviceId)
+ .forTable(TABLE_IN_PORT)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(dpPort)
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .transition(TABLE_DST_IP)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_DEFAULT)
+ .forDevice(deviceId)
+ .forTable(TABLE_IN_PORT)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+ }
+
+ private void processAccessTypeTable(DeviceId deviceId, PortNumber dpPort) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(dpPort)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_ZERO)
+ .forDevice(deviceId)
+ .forTable(TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+ }
+
+ private void processVlanTable(DeviceId deviceId, PortNumber dpPort) {
+ // for traffic going out to WAN, strip vid 500 and take through data plane interface
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchVlanId(VLAN_WAN)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .popVlan()
+ .setOutput(dpPort)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_DEFAULT)
+ .forDevice(deviceId)
+ .forTable(TABLE_VLAN)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+
+ selector = DefaultTrafficSelector.builder()
+ .matchVlanId(VLAN_WAN)
+ .matchEthType(Ethernet.TYPE_ARP)
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.CONTROLLER)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(PRIORITY_HIGH)
+ .forDevice(deviceId)
+ .forTable(TABLE_VLAN)
+ .makePermanent()
+ .build();
+
+ processFlowRule(true, flowRule);
+ }
+
+ public void processFlowRule(boolean install, FlowRule rule) {
+ FlowRuleOperations.Builder oBuilder = FlowRuleOperations.builder();
+ oBuilder = install ? oBuilder.add(rule) : oBuilder.remove(rule);
+
+ flowRuleService.apply(oBuilder.build(new FlowRuleOperationsContext() {
+ @Override
+ public void onError(FlowRuleOperations ops) {
+ log.error(String.format("Failed %s, %s", ops.toString(), rule.toString()));
+ }
+ }));
+ }
+
+ public ExtensionTreatment tunnelDstTreatment(DeviceId deviceId, Ip4Address remoteIp) {
+ try {
+ Device device = deviceService.getDevice(deviceId);
+
+ if (device.is(ExtensionTreatmentResolver.class)) {
+ ExtensionTreatmentResolver resolver = device.as(ExtensionTreatmentResolver.class);
+ ExtensionTreatment treatment =
+ resolver.getExtensionInstruction(NICIRA_SET_TUNNEL_DST.type());
+ treatment.setPropertyValue("tunnelDst", remoteIp);
+ return treatment;
+ } else {
+ log.warn("The extension treatment resolving behaviour is not supported in device {}",
+ device.id().toString());
+ return null;
+ }
+ } catch (ItemNotFoundException | UnsupportedOperationException |
+ ExtensionPropertyException e) {
+ log.error("Failed to get extension instruction {}", deviceId);
+ return null;
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/RemoteIpCommandUtil.java b/src/main/java/org/opencord/cordvtn/impl/RemoteIpCommandUtil.java
new file mode 100644
index 0000000..bd2d922
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/RemoteIpCommandUtil.java
@@ -0,0 +1,243 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl;
+
+import com.google.common.collect.Sets;
+import com.google.common.io.CharStreams;
+import com.jcraft.jsch.Channel;
+import com.jcraft.jsch.ChannelExec;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import org.onlab.packet.IpAddress;
+import org.opencord.cordvtn.api.NetworkAddress;
+import org.opencord.cordvtn.api.SshAccessInfo;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.Set;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+/**
+ * {@code RemoteIpCommandUtil} provides methods to help execute Linux IP commands to a remote server.
+ * It opens individual exec channels for each command. User can create a session with {@code connect}
+ * method and then execute a series commands. After done with all commands, the session must be closed
+ * explicitly by calling {@code disconnect}.
+ */
+public final class RemoteIpCommandUtil {
+
+ protected static final Logger log = getLogger(RemoteIpCommandUtil.class);
+
+ private static final String STRICT_HOST_CHECKING = "StrictHostKeyChecking";
+ private static final String DEFAULT_STRICT_HOST_CHECKING = "no";
+ private static final int DEFAULT_SESSION_TIMEOUT = 60000; // milliseconds
+
+ private static final String IP_PATTERN = "^([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
+ "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
+ "([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\." +
+ "([01]?\\d\\d?|2[0-4]\\d|25[0-5])$";
+
+ private static final String IP_ADDR_SHOW = "sudo ip addr show %s";
+ private static final String IP_ADDR_FLUSH = "sudo ip addr flush %s";
+ private static final String IP_ADDR_ADD = "sudo ip addr add %s dev %s";
+ private static final String IP_ADDR_DELETE = "sudo ip addr delete %s dev %s";
+ private static final String IP_LINK_SHOW = "sudo ip link show %s";
+ private static final String IP_LINK_UP = "sudo ip link set %s up";
+
+ /**
+ * Default constructor.
+ */
+ private RemoteIpCommandUtil() {
+ }
+
+ /**
+ * Adds a given IP address to a given device.
+ *
+ * @param session ssh connection
+ * @param ip network address
+ * @param device device name to assign the ip address
+ * @return true if the command succeeds, or false
+ */
+ public static boolean addIp(Session session, NetworkAddress ip, String device) {
+ if (session == null || !session.isConnected()) {
+ return false;
+ }
+
+ executeCommand(session, String.format(IP_ADDR_ADD, ip.cidr(), device));
+ Set<IpAddress> result = getCurrentIps(session, device);
+ return result.contains(ip.ip());
+ }
+
+ /**
+ * Removes the IP address from a given device.
+ *
+ * @param session ssh connection
+ * @param ip ip address
+ * @param device device name
+ * @return true if the command succeeds, or false
+ */
+ public static boolean deleteIp(Session session, IpAddress ip, String device) {
+ if (session == null || !session.isConnected()) {
+ return false;
+ }
+
+ executeCommand(session, String.format(IP_ADDR_DELETE, ip, device));
+ Set<IpAddress> result = getCurrentIps(session, device);
+ return !result.contains(ip);
+ }
+
+ /**
+ * Removes all IP address on a given device.
+ *
+ * @param session ssh connection
+ * @param device device name
+ * @return true if the command succeeds, or false
+ */
+ public static boolean flushIp(Session session, String device) {
+ if (session == null || !session.isConnected()) {
+ return false;
+ }
+
+ executeCommand(session, String.format(IP_ADDR_FLUSH, device));
+ return getCurrentIps(session, device).isEmpty();
+ }
+
+ /**
+ * Returns a set of IP address that a given device has.
+ *
+ * @param session ssh connection
+ * @param device device name
+ * @return set of IP prefix or empty set
+ */
+ public static Set<IpAddress> getCurrentIps(Session session, String device) {
+ if (session == null || !session.isConnected()) {
+ return Sets.newHashSet();
+ }
+
+ String output = executeCommand(session, String.format(IP_ADDR_SHOW, device));
+ Set<IpAddress> result = Pattern.compile(" |/")
+ .splitAsStream(output)
+ .filter(s -> s.matches(IP_PATTERN))
+ .map(IpAddress::valueOf)
+ .collect(Collectors.toSet());
+
+ return result;
+ }
+
+ /**
+ * Sets link state up for a given device.
+ *
+ * @param session ssh connection
+ * @param device device name
+ * @return true if the command succeeds, or false
+ */
+ public static boolean setInterfaceUp(Session session, String device) {
+ if (session == null || !session.isConnected()) {
+ return false;
+ }
+
+ executeCommand(session, String.format(IP_LINK_UP, device));
+ return isInterfaceUp(session, device);
+ }
+
+ /**
+ * Checks if a given interface is up or not.
+ *
+ * @param session ssh connection
+ * @param device device name
+ * @return true if the interface is up, or false
+ */
+ public static boolean isInterfaceUp(Session session, String device) {
+ if (session == null || !session.isConnected()) {
+ return false;
+ }
+
+ String output = executeCommand(session, String.format(IP_LINK_SHOW, device));
+ return output != null && output.contains("UP");
+ }
+
+ /**
+ * Creates a new session with a given access information.
+ *
+ * @param sshInfo information to ssh to the remove server
+ * @return ssh session, or null
+ */
+ public static Session connect(SshAccessInfo sshInfo) {
+ try {
+ JSch jsch = new JSch();
+ jsch.addIdentity(sshInfo.privateKey());
+
+ Session session = jsch.getSession(sshInfo.user(),
+ sshInfo.remoteIp().toString(),
+ sshInfo.port().toInt());
+ session.setConfig(STRICT_HOST_CHECKING, DEFAULT_STRICT_HOST_CHECKING);
+ session.connect(DEFAULT_SESSION_TIMEOUT);
+
+ return session;
+ } catch (JSchException e) {
+ log.debug("Failed to connect to {} due to {}", sshInfo.toString(), e.toString());
+ return null;
+ }
+ }
+
+ /**
+ * Closes a connection.
+ *
+ * @param session session
+ */
+ public static void disconnect(Session session) {
+ if (session.isConnected()) {
+ session.disconnect();
+ }
+ }
+
+ /**
+ * Executes a given command. It opens exec channel for the command and closes
+ * the channel when it's done.
+ *
+ * @param session ssh connection to a remote server
+ * @param command command to execute
+ * @return command output string if the command succeeds, or null
+ */
+ private static String executeCommand(Session session, String command) {
+ if (session == null || !session.isConnected()) {
+ return null;
+ }
+
+ log.trace("Execute command {} to {}", command, session.getHost());
+
+ try {
+ Channel channel = session.openChannel("exec");
+ ((ChannelExec) channel).setCommand(command);
+ channel.setInputStream(null);
+ InputStream output = channel.getInputStream();
+
+ channel.connect();
+ String result = CharStreams.toString(new InputStreamReader(output));
+ channel.disconnect();
+
+ return result;
+ } catch (JSchException | IOException e) {
+ log.debug("Failed to execute command {} due to {}", command, e.toString());
+ return null;
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/package-info.java b/src/main/java/org/opencord/cordvtn/impl/package-info.java
new file mode 100644
index 0000000..fe7554d
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Implementation for CORD VTN application.
+ */
+package org.opencord.cordvtn.impl;
\ No newline at end of file
diff --git a/src/main/java/org/opencord/cordvtn/impl/service/DummyInstanceHandler.java b/src/main/java/org/opencord/cordvtn/impl/service/DummyInstanceHandler.java
new file mode 100644
index 0000000..86a5f1b
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/service/DummyInstanceHandler.java
@@ -0,0 +1,57 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl.service;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+
+import org.apache.felix.scr.annotations.Deactivate;
+import org.opencord.cordvtn.api.Instance;
+import org.opencord.cordvtn.api.InstanceHandler;
+import org.opencord.cordvtn.impl.CordVtnInstanceHandler;
+import org.onosproject.xosclient.api.VtnService;
+
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+
+/**
+ * Provides network connectivity for dummy service instances.
+ */
+@Component(immediate = true)
+public class DummyInstanceHandler extends CordVtnInstanceHandler implements InstanceHandler {
+
+ @Activate
+ protected void activate() {
+ serviceType = VtnService.ServiceType.DUMMY;
+ eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn-dummy", "event-handler"));
+ super.activate();
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ super.deactivate();
+ }
+
+ @Override
+ public void instanceDetected(Instance instance) {
+ super.instanceDetected(instance);
+ }
+
+ @Override
+ public void instanceRemoved(Instance instance) {
+ super.instanceRemoved(instance);
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/service/OltAgentInstanceHandler.java b/src/main/java/org/opencord/cordvtn/impl/service/OltAgentInstanceHandler.java
new file mode 100644
index 0000000..7919c5a
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/service/OltAgentInstanceHandler.java
@@ -0,0 +1,177 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl.service;
+
+import com.google.common.collect.Maps;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IpPrefix;
+
+import org.opencord.cordvtn.api.CordVtnConfig;
+import org.opencord.cordvtn.api.Instance;
+import org.opencord.cordvtn.api.InstanceHandler;
+import org.opencord.cordvtn.impl.CordVtnInstanceHandler;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.PortNumber;
+
+import org.onosproject.net.config.ConfigFactory;
+import org.onosproject.net.config.NetworkConfigEvent;
+import org.onosproject.net.config.NetworkConfigListener;
+import org.onosproject.net.config.basics.SubjectFactories;
+import org.onosproject.net.flow.DefaultFlowRule;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.xosclient.api.VtnService;
+import org.opencord.cordconfig.access.AccessAgentConfig;
+import org.opencord.cordconfig.access.AccessAgentData;
+import org.opencord.cordvtn.impl.CordVtnPipeline;
+
+import java.util.Map;
+import java.util.Set;
+
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+
+/**
+ * Provides network connectivity for OLT agent instances.
+ */
+@Component(immediate = true)
+public class OltAgentInstanceHandler extends CordVtnInstanceHandler implements InstanceHandler {
+
+ private static final Class<AccessAgentConfig> CONFIG_CLASS = AccessAgentConfig.class;
+ private ConfigFactory<DeviceId, AccessAgentConfig> configFactory =
+ new ConfigFactory<DeviceId, AccessAgentConfig>(
+ SubjectFactories.DEVICE_SUBJECT_FACTORY, CONFIG_CLASS, "accessAgent") {
+ @Override
+ public AccessAgentConfig createConfig() {
+ return new AccessAgentConfig();
+ }
+ };
+
+ private Map<DeviceId, AccessAgentData> oltAgentData = Maps.newConcurrentMap();
+ private IpPrefix mgmtIpRange = null;
+
+ @Activate
+ protected void activate() {
+ eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn-olt", "event-handler"));
+ serviceType = VtnService.ServiceType.OLT_AGENT;
+
+ configRegistry.registerConfigFactory(configFactory);
+ configListener = new InternalConfigListener();
+
+ super.activate();
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ super.deactivate();
+ }
+
+ @Override
+ public void instanceDetected(Instance instance) {
+ log.info("OLT agent instance detected {}", instance);
+
+ managementAccessRule(instance.deviceId(), true);
+ // TODO implement
+ }
+
+ @Override
+ public void instanceRemoved(Instance instance) {
+ log.info("OLT agent instance removed {}", instance);
+
+ if (getInstances(instance.serviceId()).isEmpty()) {
+ nodeManager.completeNodes().stream().forEach(node ->
+ managementAccessRule(node.intBrId(), false));
+ }
+
+ // TODO implement
+ }
+
+ private void managementAccessRule(DeviceId deviceId, boolean install) {
+ // TODO remove this rule after long term management network is done
+ if (mgmtIpRange != null) {
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(mgmtIpRange)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(PortNumber.LOCAL)
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_MANAGEMENT)
+ .forDevice(deviceId)
+ .forTable(CordVtnPipeline.TABLE_ACCESS_TYPE)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+ }
+ }
+
+ private void readAccessAgentConfig() {
+
+ Set<DeviceId> deviceSubjects = configRegistry.getSubjects(DeviceId.class, CONFIG_CLASS);
+ deviceSubjects.stream().forEach(subject -> {
+ AccessAgentConfig config = configRegistry.getConfig(subject, CONFIG_CLASS);
+ if (config != null) {
+ oltAgentData.put(subject, config.getAgent());
+ }
+ });
+ }
+
+ @Override
+ protected void readConfiguration() {
+ CordVtnConfig config = configRegistry.getConfig(appId, CordVtnConfig.class);
+ if (config == null) {
+ log.debug("No configuration found");
+ return;
+ }
+
+ osAccess = config.openstackAccess();
+ xosAccess = config.xosAccess();
+ mgmtIpRange = config.managementIpRange();
+ }
+
+ public class InternalConfigListener implements NetworkConfigListener {
+
+ @Override
+ public void event(NetworkConfigEvent event) {
+
+ switch (event.type()) {
+ case CONFIG_UPDATED:
+ case CONFIG_ADDED:
+ if (event.configClass().equals(CordVtnConfig.class)) {
+ readConfiguration();
+ } else if (event.configClass().equals(CONFIG_CLASS)) {
+ readAccessAgentConfig();
+ }
+ break;
+ default:
+ break;
+ }
+ }
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/service/VsgInstanceHandler.java b/src/main/java/org/opencord/cordvtn/impl/service/VsgInstanceHandler.java
new file mode 100644
index 0000000..46c4a01
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/service/VsgInstanceHandler.java
@@ -0,0 +1,378 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.opencord.cordvtn.impl.service;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Sets;
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Reference;
+import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.VlanId;
+import org.opencord.cordvtn.api.Instance;
+import org.opencord.cordvtn.api.InstanceHandler;
+import org.opencord.cordvtn.impl.CordVtnInstanceHandler;
+import org.opencord.cordvtn.impl.CordVtnInstanceManager;
+import org.onosproject.net.DefaultAnnotations;
+import org.onosproject.net.HostId;
+import org.onosproject.net.PortNumber;
+import org.onosproject.net.flow.DefaultFlowRule;
+import org.onosproject.net.flow.DefaultTrafficSelector;
+import org.onosproject.net.flow.DefaultTrafficTreatment;
+import org.onosproject.net.flow.FlowRule;
+import org.onosproject.net.flow.FlowRuleService;
+import org.onosproject.net.flow.TrafficSelector;
+import org.onosproject.net.flow.TrafficTreatment;
+import org.onosproject.net.flow.criteria.Criterion;
+import org.onosproject.net.flow.criteria.IPCriterion;
+import org.onosproject.net.flow.instructions.Instruction;
+import org.onosproject.net.flow.instructions.Instructions;
+import org.onosproject.net.flow.instructions.L2ModificationInstruction;
+import org.onosproject.net.host.DefaultHostDescription;
+import org.onosproject.net.host.HostDescription;
+import org.onosproject.xosclient.api.VtnPort;
+import org.onosproject.xosclient.api.VtnPortApi;
+import org.onosproject.xosclient.api.VtnPortId;
+import org.onosproject.xosclient.api.VtnService;
+import org.opencord.cordvtn.impl.CordVtnPipeline;
+
+import java.util.Map;
+import java.util.Set;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor;
+import static org.onlab.util.Tools.groupedThreads;
+import static org.onosproject.net.flow.criteria.Criterion.Type.IPV4_DST;
+import static org.onosproject.net.flow.instructions.L2ModificationInstruction.L2SubType.VLAN_PUSH;
+
+/**
+ * Provides network connectivity for vSG instances.
+ */
+@Component(immediate = true)
+@Service(value = VsgInstanceHandler.class)
+public final class VsgInstanceHandler extends CordVtnInstanceHandler implements InstanceHandler {
+
+ private static final String STAG = "stag";
+ private static final String VSG_VM = "vsgVm";
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected FlowRuleService flowRuleService;
+
+ @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
+ protected CordVtnInstanceManager instanceManager;
+
+ @Activate
+ protected void activate() {
+ serviceType = VtnService.ServiceType.VSG;
+ eventExecutor = newSingleThreadScheduledExecutor(groupedThreads("onos/cordvtn-vsg", "event-handler"));
+ super.activate();
+ }
+
+ @Deactivate
+ protected void deactivate() {
+ super.deactivate();
+ }
+
+ @Override
+ public void instanceDetected(Instance instance) {
+ if (isVsgContainer(instance)) {
+ log.info("vSG container detected {}", instance);
+
+ // find vsg vm for this vsg container
+ String vsgVmId = instance.getAnnotation(VSG_VM);
+ if (Strings.isNullOrEmpty(vsgVmId)) {
+ log.warn("Failed to find VSG VM for {}", instance);
+ return;
+ }
+
+ Instance vsgVm = Instance.of(hostService.getHost(HostId.hostId(vsgVmId)));
+ VtnPort vtnPort = getVtnPort(vsgVm);
+ if (vtnPort == null || getStag(vtnPort) == null) {
+ return;
+ }
+
+ populateVsgRules(vsgVm, getStag(vtnPort),
+ nodeManager.dpPort(vsgVm.deviceId()),
+ vtnPort.addressPairs().keySet(),
+ true);
+
+ } else {
+ VtnPort vtnPort = getVtnPort(instance);
+ if (vtnPort == null || getStag(vtnPort) == null) {
+ return;
+ }
+
+ vtnPort.addressPairs().entrySet().stream()
+ .forEach(pair -> addVsgContainer(
+ instance,
+ pair.getKey(),
+ pair.getValue(),
+ getStag(vtnPort).toString()
+ ));
+ super.instanceDetected(instance);
+ }
+ }
+
+ @Override
+ public void instanceRemoved(Instance instance) {
+ if (isVsgContainer(instance)) {
+ log.info("vSG container vanished {}", instance);
+
+ // find vsg vm for this vsg container
+ String vsgVmId = instance.getAnnotation(VSG_VM);
+ if (Strings.isNullOrEmpty(vsgVmId)) {
+ log.warn("Failed to find VSG VM for {}", instance);
+ return;
+ }
+
+ Instance vsgVm = Instance.of(hostService.getHost(HostId.hostId(vsgVmId)));
+ VtnPort vtnPort = getVtnPort(vsgVm);
+ if (vtnPort == null || getStag(vtnPort) == null) {
+ return;
+ }
+
+ populateVsgRules(vsgVm, getStag(vtnPort),
+ nodeManager.dpPort(vsgVm.deviceId()),
+ vtnPort.addressPairs().keySet(),
+ false);
+
+ } else {
+ // TODO remove vsg vm related rules
+ super.instanceRemoved(instance);
+ }
+ }
+
+ /**
+ * Updates set of vSGs in a given vSG VM.
+ *
+ * @param vsgVmId vsg vm host id
+ * @param stag stag
+ * @param vsgInstances full set of vsg wan ip and mac address pairs in this vsg vm
+ */
+ public void updateVsgInstances(HostId vsgVmId, String stag, Map<IpAddress, MacAddress> vsgInstances) {
+ if (hostService.getHost(vsgVmId) == null) {
+ log.debug("vSG VM {} is not added yet, ignore this update", vsgVmId);
+ return;
+ }
+
+ Instance vsgVm = Instance.of(hostService.getHost(vsgVmId));
+ if (vsgVm == null) {
+ log.warn("Failed to find existing vSG VM for STAG: {}", stag);
+ return;
+ }
+
+ log.info("Updates vSGs in {} with STAG: {}", vsgVm, stag);
+
+ // adds vSGs in the address pair
+ vsgInstances.entrySet().stream()
+ .filter(addr -> hostService.getHostsByMac(addr.getValue()).isEmpty())
+ .forEach(addr -> addVsgContainer(
+ vsgVm,
+ addr.getKey(),
+ addr.getValue(),
+ stag));
+
+ // removes vSGs not listed in the address pair
+ hostService.getConnectedHosts(vsgVm.host().location()).stream()
+ .filter(host -> !host.mac().equals(vsgVm.mac()))
+ .filter(host -> !vsgInstances.values().contains(host.mac()))
+ .forEach(host -> {
+ log.info("Removed vSG {}", host.toString());
+ instanceManager.removeInstance(host.id());
+ });
+ }
+
+ private boolean isVsgContainer(Instance instance) {
+ return !Strings.isNullOrEmpty(instance.host().annotations().value(STAG));
+ }
+
+ private void addVsgContainer(Instance vsgVm, IpAddress vsgWanIp, MacAddress vsgMac,
+ String stag) {
+ HostId hostId = HostId.hostId(vsgMac);
+ DefaultAnnotations.Builder annotations = DefaultAnnotations.builder()
+ .set(Instance.SERVICE_TYPE, vsgVm.serviceType().toString())
+ .set(Instance.SERVICE_ID, vsgVm.serviceId().id())
+ .set(Instance.PORT_ID, vsgVm.portId().id())
+ .set(Instance.NESTED_INSTANCE, Instance.TRUE)
+ .set(STAG, stag)
+ .set(VSG_VM, vsgVm.host().id().toString())
+ .set(Instance.CREATE_TIME, String.valueOf(System.currentTimeMillis()));
+
+ HostDescription hostDesc = new DefaultHostDescription(
+ vsgMac,
+ VlanId.NONE,
+ vsgVm.host().location(),
+ Sets.newHashSet(vsgWanIp),
+ annotations.build());
+
+ instanceManager.addInstance(hostId, hostDesc);
+ }
+
+ private void populateVsgRules(Instance vsgVm, VlanId stag, PortNumber dpPort,
+ Set<IpAddress> vsgWanIps, boolean install) {
+ // for traffics with s-tag, strip the tag and take through the vSG VM
+ TrafficSelector selector = DefaultTrafficSelector.builder()
+ .matchInPort(dpPort)
+ .matchVlanId(stag)
+ .build();
+
+ TrafficTreatment treatment = DefaultTrafficTreatment.builder()
+ .setOutput(vsgVm.portNumber())
+ .build();
+
+ FlowRule flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(vsgVm.deviceId())
+ .forTable(CordVtnPipeline.TABLE_VLAN)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ // for traffics with customer vlan, tag with the service vlan based on input port with
+ // lower priority to avoid conflict with WAN tag
+ selector = DefaultTrafficSelector.builder()
+ .matchInPort(vsgVm.portNumber())
+ .matchVlanId(stag)
+ .build();
+
+ treatment = DefaultTrafficTreatment.builder()
+ .setOutput(dpPort)
+ .build();
+
+ flowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(selector)
+ .withTreatment(treatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(vsgVm.deviceId())
+ .forTable(CordVtnPipeline.TABLE_VLAN)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, flowRule);
+
+ // for traffic coming from WAN, tag 500 and take through the vSG VM
+ // based on destination ip
+ vsgWanIps.stream().forEach(ip -> {
+ TrafficSelector downstream = DefaultTrafficSelector.builder()
+ .matchEthType(Ethernet.TYPE_IPV4)
+ .matchIPDst(ip.toIpPrefix())
+ .build();
+
+ TrafficTreatment downstreamTreatment = DefaultTrafficTreatment.builder()
+ .pushVlan()
+ .setVlanId(CordVtnPipeline.VLAN_WAN)
+ .setEthDst(vsgVm.mac())
+ .setOutput(vsgVm.portNumber())
+ .build();
+
+ FlowRule downstreamFlowRule = DefaultFlowRule.builder()
+ .fromApp(appId)
+ .withSelector(downstream)
+ .withTreatment(downstreamTreatment)
+ .withPriority(CordVtnPipeline.PRIORITY_DEFAULT)
+ .forDevice(vsgVm.deviceId())
+ .forTable(CordVtnPipeline.TABLE_DST_IP)
+ .makePermanent()
+ .build();
+
+ pipeline.processFlowRule(install, downstreamFlowRule);
+ });
+
+ // remove downstream flow rules for the vSG not shown in vsgWanIps
+ for (FlowRule rule : flowRuleService.getFlowRulesById(appId)) {
+ if (!rule.deviceId().equals(vsgVm.deviceId())) {
+ continue;
+ }
+ PortNumber output = getOutputFromTreatment(rule);
+ if (output == null || !output.equals(vsgVm.portNumber()) ||
+ !isVlanPushFromTreatment(rule)) {
+ continue;
+ }
+
+ IpPrefix dstIp = getDstIpFromSelector(rule);
+ if (dstIp != null && !vsgWanIps.contains(dstIp.address())) {
+ pipeline.processFlowRule(false, rule);
+ }
+ }
+ }
+
+ private VtnPort getVtnPort(Instance instance) {
+ checkNotNull(osAccess, OPENSTACK_ACCESS_ERROR);
+ checkNotNull(xosAccess, XOS_ACCESS_ERROR);
+
+ VtnPortId vtnPortId = instance.portId();
+ VtnPortApi portApi = xosClient.getClient(xosAccess).vtnPort();
+ VtnPort vtnPort = portApi.vtnPort(vtnPortId, osAccess);
+ if (vtnPort == null) {
+ log.warn("Failed to get port information of {}", instance);
+ return null;
+ }
+ return vtnPort;
+ }
+
+ // TODO get stag from XOS when XOS provides it, extract if from port name for now
+ private VlanId getStag(VtnPort vtnPort) {
+ checkNotNull(vtnPort);
+
+ String portName = vtnPort.name();
+ if (portName != null && portName.startsWith(STAG)) {
+ return VlanId.vlanId(portName.split("-")[1]);
+ } else {
+ return null;
+ }
+ }
+
+ private PortNumber getOutputFromTreatment(FlowRule flowRule) {
+ Instruction instruction = flowRule.treatment().allInstructions().stream()
+ .filter(inst -> inst instanceof Instructions.OutputInstruction)
+ .findFirst()
+ .orElse(null);
+ if (instruction == null) {
+ return null;
+ }
+ return ((Instructions.OutputInstruction) instruction).port();
+ }
+
+ private IpPrefix getDstIpFromSelector(FlowRule flowRule) {
+ Criterion criterion = flowRule.selector().getCriterion(IPV4_DST);
+ if (criterion != null && criterion instanceof IPCriterion) {
+ IPCriterion ip = (IPCriterion) criterion;
+ return ip.ip();
+ } else {
+ return null;
+ }
+ }
+
+ private boolean isVlanPushFromTreatment(FlowRule flowRule) {
+ Instruction instruction = flowRule.treatment().allInstructions().stream()
+ .filter(inst -> inst instanceof L2ModificationInstruction)
+ .filter(inst -> ((L2ModificationInstruction) inst).subtype().equals(VLAN_PUSH))
+ .findAny()
+ .orElse(null);
+ return instruction != null;
+ }
+}
diff --git a/src/main/java/org/opencord/cordvtn/impl/service/package-info.java b/src/main/java/org/opencord/cordvtn/impl/service/package-info.java
new file mode 100644
index 0000000..7d00d9f
--- /dev/null
+++ b/src/main/java/org/opencord/cordvtn/impl/service/package-info.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Implementation of instance handlers for various network services.
+ */
+package org.opencord.cordvtn.impl.service;
\ No newline at end of file