/* * Copyright (C) 2013 Red Hat, Inc. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html * * Authors : Madhu Venugopal, Brent Salisbury, Sam Hague, Dave Tucker */ package org.opendaylight.ovsdb.neutron; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; import org.opendaylight.controller.networkconfig.neutron.NeutronNetwork; import org.opendaylight.controller.sal.core.Node; import org.opendaylight.controller.sal.core.NodeConnector; import org.opendaylight.controller.sal.core.Property; import org.opendaylight.controller.sal.core.UpdateType; import org.opendaylight.controller.switchmanager.IInventoryListener; import org.opendaylight.ovsdb.lib.notation.UUID; import org.opendaylight.ovsdb.lib.table.Interface; import org.opendaylight.ovsdb.lib.table.Open_vSwitch; import org.opendaylight.ovsdb.lib.table.Port; import org.opendaylight.ovsdb.lib.table.Table; import org.opendaylight.ovsdb.neutron.provider.IProviderNetworkManager; import org.opendaylight.ovsdb.plugin.OVSDBInventoryListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; public class SouthboundHandler extends BaseHandler implements OVSDBInventoryListener, IInventoryListener { static final Logger logger = LoggerFactory.getLogger(SouthboundHandler.class); //private Thread eventThread; private ExecutorService eventHandler; private BlockingQueue events; List nodeCache; // The implementation for each of these services is resolved by the OSGi Service Manager private volatile IAdminConfigManager adminConfigManager; private volatile IInternalNetworkManager internalNetworkManager; private volatile ITenantNetworkManager tenantNetworkManager; private volatile IProviderNetworkManager providerNetworkManager; void init() { eventHandler = Executors.newSingleThreadExecutor(); this.events = new LinkedBlockingQueue(); nodeCache = new ArrayList<>(); } void start() { eventHandler.submit(new Runnable() { @Override public void run() { while (true) { SouthboundEvent ev; try { ev = events.take(); } catch (InterruptedException e) { logger.info("The event handler thread was interrupted, shutting down", e); return; } switch (ev.getType()) { case NODE: try { processNodeUpdate(ev.getNode(), ev.getAction()); } catch (Exception e) { logger.error("Exception caught in ProcessNodeUpdate for node " + ev.getNode(), e); } break; case ROW: try { processRowUpdate(ev.getNode(), ev.getTableName(), ev.getUuid(), ev.getRow(), ev.getContext(),ev.getAction()); } catch (Exception e) { logger.error("Exception caught in ProcessRowUpdate for node " + ev.getNode(), e); } break; default: logger.warn("Unable to process action " + ev.getAction() + " for node " + ev.getNode()); } } } }); this.triggerUpdates(); } void stop() { eventHandler.shutdownNow(); } @Override public void nodeAdded(Node node) { this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.ADD)); } @Override public void nodeRemoved(Node node) { this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.DELETE)); } @Override public void rowAdded(Node node, String tableName, String uuid, Table row) { this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.ADD)); } @Override public void rowUpdated(Node node, String tableName, String uuid, Table oldRow, Table newRow) { if (this.isUpdateOfInterest(node, oldRow, newRow)) { this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, newRow, SouthboundEvent.Action.UPDATE)); } } /* * Ignore unneccesary updates to be even considered for processing. * (Especially stats update are fast and furious). */ private boolean isUpdateOfInterest(Node node, Table oldRow, Table newRow) { if (oldRow == null) return true; if (newRow.getTableName().equals(Interface.NAME)) { // We are NOT interested in Stats only updates Interface oldIntf = (Interface)oldRow; if (oldIntf.getName() == null && oldIntf.getExternal_ids() == null && oldIntf.getMac() == null && oldIntf.getOfport() == null && oldIntf.getOptions() == null && oldIntf.getOther_config() == null && oldIntf.getType() == null) { logger.trace("IGNORING Interface Update: node {}, row: {}", node, newRow); return false; } } else if (newRow.getTableName().equals(Port.NAME)) { // We are NOT interested in Stats only updates Port oldPort = (Port)oldRow; if (oldPort.getName() == null && oldPort.getExternal_ids() == null && oldPort.getMac() == null && oldPort.getInterfaces() == null && oldPort.getTag() == null && oldPort.getTrunks() == null) { logger.trace("IGNORING Port Update: node {}, row: {}", node, newRow); return false; } } else if (newRow.getTableName().equals(Open_vSwitch.NAME)) { Open_vSwitch oldOpenvSwitch = (Open_vSwitch) oldRow; if (oldOpenvSwitch.getOther_config()== null) { /* we are only interested in other_config field change */ return false; } } return true; } @Override public void rowRemoved(Node node, String tableName, String uuid, Table row, Object context) { this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, context, SouthboundEvent.Action.DELETE)); } private void enqueueEvent (SouthboundEvent event) { try { events.put(event); } catch (InterruptedException e) { logger.error("Thread was interrupted while trying to enqueue event ", e); } } public void processNodeUpdate(Node node, SouthboundEvent.Action action) { if (action == SouthboundEvent.Action.DELETE) return; logger.trace("Process Node added {}", node); internalNetworkManager.prepareInternalNetwork(node); } private void processRowUpdate(Node node, String tableName, String uuid, Table row, Object context, SouthboundEvent.Action action) { if (action == SouthboundEvent.Action.DELETE) { if (Interface.NAME.getName().equalsIgnoreCase(tableName)) { logger.debug("processRowUpdate: {} Deleted node: {}, uuid: {}, row: {}", tableName, node, uuid, row); Interface deletedIntf = (Interface)row; NeutronNetwork network = null; if (context == null) { network = tenantNetworkManager.getTenantNetworkForInterface(deletedIntf); } else { network = (NeutronNetwork)context; } List phyIfName = adminConfigManager.getAllPhysicalInterfaceNames(node); logger.info("Delete interface " + deletedIntf.getName()); if (deletedIntf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) || deletedIntf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) || phyIfName.contains(deletedIntf.getName())) { /* delete tunnel interfaces or physical interfaces */ this.handleInterfaceDelete(node, uuid, deletedIntf, false, null); } else if (network != null && !network.getRouterExternal()) { try { ConcurrentMap> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName()); if (interfaces != null) { boolean isLastInstanceOnNode = true; for (String intfUUID : interfaces.keySet()) { if (intfUUID.equals(uuid)) continue; Interface intf = (Interface) interfaces.get(intfUUID); NeutronNetwork neutronNetwork = tenantNetworkManager.getTenantNetworkForInterface(intf); if (neutronNetwork != null && neutronNetwork.equals(network)) isLastInstanceOnNode = false; } this.handleInterfaceDelete(node, uuid, deletedIntf, isLastInstanceOnNode, network); } } catch (Exception e) { logger.error("Error fetching Interface Rows for node " + node, e); } } } } else if (Interface.NAME.getName().equalsIgnoreCase(tableName)) { logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row); Interface intf = (Interface)row; NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf); if (network != null && !network.getRouterExternal()) { if (providerNetworkManager.getProvider().hasPerTenantTunneling()) { int vlan = tenantNetworkManager.networkCreated(node, network.getID()); logger.trace("Neutron Network {}:{} Created with Internal Vlan: {}", network.getNetworkUUID(), network.getNetworkName(), vlan); String portUUID = this.getPortIdForInterface(node, uuid, intf); if (portUUID != null) { tenantNetworkManager.programTenantNetworkInternalVlan(node, portUUID, network); } } this.handleInterfaceUpdate(node, uuid, intf); } } else if (Port.NAME.getName().equalsIgnoreCase(tableName)) { logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row); Port port = (Port)row; Set interfaceUUIDs = port.getInterfaces(); for (UUID intfUUID : interfaceUUIDs) { logger.trace("Scanning interface "+intfUUID); try { Interface intf = (Interface)this.ovsdbConfigService.getRow(node, Interface.NAME.getName(), intfUUID.toString()); NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf); if (network != null && !network.getRouterExternal()) { tenantNetworkManager.programTenantNetworkInternalVlan(node, uuid, network); } else { logger.trace("ignore update because there is not a neutron network."); } } catch (Exception e) { logger.error("Failed to process row update", e); } } } else if (Open_vSwitch.NAME.getName().equalsIgnoreCase(tableName)) { logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row); try { ConcurrentMap> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName()); if (interfaces != null) { for (String intfUUID : interfaces.keySet()) { Interface intf = (Interface) interfaces.get(intfUUID); this.handleInterfaceUpdate(node, intfUUID, intf); } } } catch (Exception e) { logger.error("Error fetching Interface Rows for node " + node, e); } } } private void handleInterfaceUpdate (Node node, String uuid, Interface intf) { logger.debug("handleInterfaceUpdate: node: {}, uuid: {}", node, uuid); NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf); if (network != null) { if (internalNetworkManager.checkAndCreateNetwork(node, network)) { providerNetworkManager.getProvider().handleInterfaceUpdate(network, node, intf); } } } private void handleInterfaceDelete (Node node, String uuid, Interface intf, boolean isLastInstanceOnNode, NeutronNetwork network) { logger.debug("handleInterfaceDelete: node: {}, uuid: {}, isLastInstanceOnNode: {}, interface: {}", node, uuid, isLastInstanceOnNode, intf); List phyIfName = adminConfigManager.getAllPhysicalInterfaceNames(node); if (intf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) || intf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) || phyIfName.contains(intf.getName())) { /* delete tunnel or physical interfaces */ providerNetworkManager.getProvider().handleInterfaceDelete(intf.getType(), null, node, intf, isLastInstanceOnNode); } else if (network != null) { if (!network.getProviderNetworkType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VLAN)) { /* vlan doesn't need a tunnel endpoint */ if (adminConfigManager.getTunnelEndPoint(node) == null) { logger.error("Tunnel end-point configuration missing. Please configure it in Open_vSwitch Table"); return; } } if (isLastInstanceOnNode & providerNetworkManager.getProvider().hasPerTenantTunneling()) { tenantNetworkManager.reclaimTenantNetworkInternalVlan(node, uuid, network); } providerNetworkManager.getProvider().handleInterfaceDelete(network.getProviderNetworkType(), network, node, intf, isLastInstanceOnNode); } } private String getPortIdForInterface (Node node, String uuid, Interface intf) { try { Map> ports = this.ovsdbConfigService.getRows(node, Port.NAME.getName()); if (ports == null) return null; for (String portUUID : ports.keySet()) { Port port = (Port)ports.get(portUUID); Set interfaceUUIDs = port.getInterfaces(); logger.trace("Scanning Port {} to identify interface : {} ",port, uuid); for (UUID intfUUID : interfaceUUIDs) { if (intfUUID.toString().equalsIgnoreCase(uuid)) { logger.trace("Found Interface {} -> {}", uuid, portUUID); return portUUID; } } } } catch (Exception e) { logger.debug("Failed to add Port tag for for Intf {}",intf, e); } return null; } @Override public void notifyNode(Node node, UpdateType type, Map propMap) { logger.debug("notifyNode: Node {} update {} from Controller's inventory Service", node, type); // Add the Node Type check back once the Consistency issue is resolved between MD-SAL and AD-SAL if (!type.equals(UpdateType.REMOVED) && !nodeCache.contains(node)) { nodeCache.add(node); providerNetworkManager.getProvider().initializeOFFlowRules(node); } else if (type.equals(UpdateType.REMOVED)){ nodeCache.remove(node); } } @Override public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map propMap) { //We are not interested in the nodeConnectors at this moment } private void triggerUpdates() { List nodes = this.getConnectionService().getNodes(); if (nodes == null) return; for (Node node : nodes) { try { List tableNames = this.getOVSDBConfigService().getTables(node); if (tableNames == null) continue; for (String tableName : tableNames) { Map> rows = this.getOVSDBConfigService().getRows(node, tableName); if (rows == null) continue; for (String uuid : rows.keySet()) { Table row = rows.get(uuid); this.rowAdded(node, tableName, uuid, row); } } } catch (Exception e) { logger.error("Exception during OVSDB Southbound update trigger", e); } } } }