openstack/net-virt/src/main/java/org/opendaylight/ovsdb/openstack/netvirt/SouthboundHandler.java

   1 /*
   2  * Copyright (C) 2013 Red Hat, Inc.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  *
   8  * Authors : Madhu Venugopal, Brent Salisbury, Sam Hague, Dave Tucker
   9  */
  10 package org.opendaylight.ovsdb.openstack.netvirt;
  11
  12 import org.opendaylight.controller.networkconfig.neutron.NeutronNetwork;
  13 import org.opendaylight.controller.sal.core.Node;
  14 import org.opendaylight.controller.sal.core.NodeConnector;
  15 import org.opendaylight.controller.sal.core.Property;
  16 import org.opendaylight.controller.sal.core.UpdateType;
  17 import org.opendaylight.controller.switchmanager.IInventoryListener;
  18 import org.opendaylight.ovsdb.lib.notation.Row;
  19 import org.opendaylight.ovsdb.lib.notation.UUID;
  20 import org.opendaylight.ovsdb.openstack.netvirt.api.BridgeConfigurationManager;
  21 import org.opendaylight.ovsdb.openstack.netvirt.api.NetworkingProviderManager;
  22 import org.opendaylight.ovsdb.openstack.netvirt.api.TenantNetworkManager;
  23 import org.opendaylight.ovsdb.plugin.api.OvsdbConfigurationService;
  24 import org.opendaylight.ovsdb.plugin.api.OvsdbConnectionService;
  25 import org.opendaylight.ovsdb.plugin.api.OvsdbInventoryListener;
  26 import org.opendaylight.ovsdb.schema.openvswitch.Interface;
  27 import org.opendaylight.ovsdb.schema.openvswitch.OpenVSwitch;
  28 import org.opendaylight.ovsdb.schema.openvswitch.Port;
  29
  30 import com.google.common.collect.Lists;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33
  34 import java.net.InetAddress;
  35 import java.util.List;
  36 import java.util.Map;
  37 import java.util.Set;
  38 import java.util.concurrent.BlockingQueue;
  39 import java.util.concurrent.ConcurrentMap;
  40 import java.util.concurrent.ExecutorService;
  41 import java.util.concurrent.Executors;
  42 import java.util.concurrent.LinkedBlockingQueue;
  43
  44 public class SouthboundHandler extends AbstractHandler implements OvsdbInventoryListener, IInventoryListener {
  45     static final Logger logger = LoggerFactory.getLogger(SouthboundHandler.class);
  46     //private Thread eventThread;
  47     private ExecutorService eventHandler;
  48     private BlockingQueue<SouthboundEvent> events;
  49     List<Node> nodeCache;
  50
  51     // The implementation for each of these services is resolved by the OSGi Service Manager
  52     private volatile org.opendaylight.ovsdb.openstack.netvirt.api.ConfigurationService configurationService;
  53     private volatile BridgeConfigurationManager bridgeConfigurationManager;
  54     private volatile TenantNetworkManager tenantNetworkManager;
  55     private volatile NetworkingProviderManager networkingProviderManager;
  56     private volatile OvsdbConfigurationService ovsdbConfigurationService;
  57     private volatile OvsdbConnectionService connectionService;
  58
  59     void init() {
  60         eventHandler = Executors.newSingleThreadExecutor();
  61         this.events = new LinkedBlockingQueue<>();
  62         nodeCache = Lists.newArrayList();
  63     }
  64
  65     void start() {
  66         eventHandler.submit(new Runnable()  {
  67             @Override
  68             public void run() {
  69                 while (true) {
  70                     SouthboundEvent ev;
  71                     try {
  72                         ev = events.take();
  73                     } catch (InterruptedException e) {
  74                         logger.info("The event handler thread was interrupted, shutting down", e);
  75                         return;
  76                     }
  77                     switch (ev.getType()) {
  78                     case NODE:
  79                         try {
  80                             processNodeUpdate(ev.getNode(), ev.getAction());
  81                         } catch (Exception e) {
  82                             logger.error("Exception caught in ProcessNodeUpdate for node " + ev.getNode(), e);
  83                         }
  84                         break;
  85                     case ROW:
  86                         try {
  87                             processRowUpdate(ev.getNode(), ev.getTableName(), ev.getUuid(), ev.getRow(),
  88                                              ev.getContext(),ev.getAction());
  89                         } catch (Exception e) {
  90                             logger.error("Exception caught in ProcessRowUpdate for node " + ev.getNode(), e);
  91                         }
  92                         break;
  93                     default:
  94                         logger.warn("Unable to process action " + ev.getAction() + " for node " + ev.getNode());
  95                     }
  96                 }
  97             }
  98         });
  99         this.triggerUpdates();
 100     }
 101
 102     void stop() {
 103         eventHandler.shutdownNow();
 104     }
 105
 106     @Override
 107     public void nodeAdded(Node node, InetAddress address, int port) {
 108         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.ADD));
 109     }
 110
 111     @Override
 112     public void nodeRemoved(Node node) {
 113         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.DELETE));
 114     }
 115
 116     @Override
 117     public void rowAdded(Node node, String tableName, String uuid, Row row) {
 118         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.ADD));
 119     }
 120
 121     @Override
 122     public void rowUpdated(Node node, String tableName, String uuid, Row oldRow, Row newRow) {
 123         if (this.isUpdateOfInterest(node, oldRow, newRow)) {
 124             this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, newRow, SouthboundEvent.Action.UPDATE));
 125         }
 126     }
 127
 128     /*
 129      * Ignore unneccesary updates to be even considered for processing.
 130      * (Especially stats update are fast and furious).
 131      */
 132
 133     private boolean isUpdateOfInterest(Node node, Row oldRow, Row newRow) {
 134         if (oldRow == null) return true;
 135         if (newRow.getTableSchema().getName().equals(ovsdbConfigurationService.getTableName(node, Interface.class))) {
 136             // We are NOT interested in Stats only updates
 137             Interface oldIntf = ovsdbConfigurationService.getTypedRow(node, Interface.class, oldRow);
 138             if (oldIntf.getName() == null && oldIntf.getExternalIdsColumn() == null && oldIntf.getMacColumn() == null &&
 139                 oldIntf.getOpenFlowPortColumn() == null && oldIntf.getOptionsColumn() == null && oldIntf.getOtherConfigColumn() == null &&
 140                 oldIntf.getTypeColumn() == null) {
 141                 logger.trace("IGNORING Interface Update: node {}, row: {}", node, newRow);
 142                 return false;
 143             }
 144         } else if (newRow.getTableSchema().getName().equals(ovsdbConfigurationService.getTableName(node, Port.class))) {
 145             // We are NOT interested in Stats only updates
 146             Port oldPort = ovsdbConfigurationService.getTypedRow(node, Port.class, oldRow);
 147             if (oldPort.getName() == null && oldPort.getExternalIdsColumn() == null && oldPort.getMacColumn() == null &&
 148                 oldPort.getInterfacesColumn() == null && oldPort.getTagColumn() == null && oldPort.getTrunksColumn() == null) {
 149                 logger.trace("IGNORING Port Update: node {}, row: {}", node, newRow);
 150                 return false;
 151             }
 152         } else if (newRow.getTableSchema().getName().equals(ovsdbConfigurationService.getTableName(node, OpenVSwitch.class))) {
 153             OpenVSwitch oldOpenvSwitch = ovsdbConfigurationService.getTypedRow(node, OpenVSwitch.class, oldRow);
 154             if (oldOpenvSwitch.getOtherConfigColumn()== null) {
 155                 /* we are only interested in other_config field change */
 156                 return false;
 157             }
 158         }
 159         return true;
 160     }
 161
 162     @Override
 163     public void rowRemoved(Node node, String tableName, String uuid, Row row, Object context) {
 164         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, context, SouthboundEvent.Action.DELETE));
 165     }
 166
 167     private void enqueueEvent (SouthboundEvent event) {
 168         try {
 169             events.put(event);
 170         } catch (InterruptedException e) {
 171             logger.error("Thread was interrupted while trying to enqueue event ", e);
 172         }
 173     }
 174
 175     public void processNodeUpdate(Node node, SouthboundEvent.Action action) {
 176         if (action == SouthboundEvent.Action.DELETE) return;
 177         logger.trace("Process Node added {}", node);
 178         bridgeConfigurationManager.prepareNode(node);
 179     }
 180
 181     private void processRowUpdate(Node node, String tableName, String uuid, Row row,
 182                                   Object context, SouthboundEvent.Action action) {
 183         if (action == SouthboundEvent.Action.DELETE) {
 184             if (tableName.equalsIgnoreCase(ovsdbConfigurationService.getTableName(node, Interface.class))) {
 185                 logger.debug("Processing update of {}. Deleted node: {}, uuid: {}, row: {}", tableName, node, uuid, row);
 186                 Interface deletedIntf = ovsdbConfigurationService.getTypedRow(node, Interface.class, row);
 187                 NeutronNetwork network = null;
 188                 if (context == null) {
 189                     network = tenantNetworkManager.getTenantNetwork(deletedIntf);
 190                 } else {
 191                     network = (NeutronNetwork)context;
 192                 }
 193                 List<String> phyIfName = bridgeConfigurationManager.getAllPhysicalInterfaceNames(node);
 194                 logger.info("Delete interface " + deletedIntf.getName());
 195
 196                 if (deletedIntf.getTypeColumn().getData().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) ||
 197                     deletedIntf.getTypeColumn().getData().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) ||
 198                     phyIfName.contains(deletedIntf.getName())) {
 199                     /* delete tunnel interfaces or physical interfaces */
 200                     this.handleInterfaceDelete(node, uuid, deletedIntf, false, null);
 201                 } else if (network != null && !network.getRouterExternal()) {
 202                     logger.debug("Processing update of {}:{} node {} intf {} network {}",
 203                             tableName, action, node, uuid, network.getNetworkUUID());
 204                     try {
 205                         ConcurrentMap<String, Row> interfaces = this.ovsdbConfigurationService
 206                                 .getRows(node, ovsdbConfigurationService.getTableName(node, Interface.class));
 207                         if (interfaces != null) {
 208                             boolean isLastInstanceOnNode = true;
 209                             for (String intfUUID : interfaces.keySet()) {
 210                                 if (intfUUID.equals(uuid)) continue;
 211                                 Interface intf = this.ovsdbConfigurationService.getTypedRow(node, Interface.class, interfaces.get(intfUUID));
 212                                 NeutronNetwork neutronNetwork = tenantNetworkManager.getTenantNetwork(intf);
 213                                 if (neutronNetwork != null && neutronNetwork.equals(network)) isLastInstanceOnNode = false;
 214                             }
 215                             this.handleInterfaceDelete(node, uuid, deletedIntf, isLastInstanceOnNode, network);
 216                         }
 217                     } catch (Exception e) {
 218                         logger.error("Error fetching Interface Rows for node " + node, e);
 219                     }
 220                 }
 221             }
 222         }
 223         else if (tableName.equalsIgnoreCase(ovsdbConfigurationService.getTableName(node, Interface.class))) {
 224             logger.debug("Processing update of {}:{} node: {}, interface uuid: {}, row: {}",
 225                     tableName, action, node, uuid, row);
 226             Interface intf = this.ovsdbConfigurationService.getTypedRow(node, Interface.class, row);
 227             NeutronNetwork network = tenantNetworkManager.getTenantNetwork(intf);
 228             if (network != null && !network.getRouterExternal()) {
 229                 if (networkingProviderManager.getProvider(node).hasPerTenantTunneling()) {
 230                     int vlan = tenantNetworkManager.networkCreated(node, network.getID());
 231                     String portUUID = this.getPortIdForInterface(node, uuid, intf);
 232                     if (portUUID != null) {
 233                         logger.debug("Neutron Network {}:{} Created with Internal vlan {} port {}",
 234                                  network.getNetworkUUID(), network.getNetworkName(), vlan, portUUID);
 235                         tenantNetworkManager.programInternalVlan(node, portUUID, network);
 236                     } else {
 237                         logger.trace("Neutron Network {}:{} Created with Internal vlan {} but have no portUUID",
 238                                  network.getNetworkUUID(), network.getNetworkName(), vlan);
 239                     }
 240                 }
 241                 this.handleInterfaceUpdate(node, uuid, intf);
 242             }
 243
 244         } else if (tableName.equalsIgnoreCase(ovsdbConfigurationService.getTableName(node, Port.class))) {
 245             logger.debug("Processing update of {}:{} node: {}, port uuid: {}, row: {}", tableName, action, node, uuid, row);
 246             Port port = this.ovsdbConfigurationService.getTypedRow(node, Port.class, row);
 247             Set<UUID> interfaceUUIDs = port.getInterfacesColumn().getData();
 248             for (UUID intfUUID : interfaceUUIDs) {
 249                 logger.trace("Scanning interface "+intfUUID);
 250                 try {
 251                     Row intfRow = this.ovsdbConfigurationService
 252                             .getRow(node, ovsdbConfigurationService.getTableName(node, Interface.class),
 253                                     intfUUID.toString());
 254                     Interface intf = this.ovsdbConfigurationService.getTypedRow(node, Interface.class, intfRow);
 255                     NeutronNetwork network = tenantNetworkManager.getTenantNetwork(intf);
 256                     if (network != null && !network.getRouterExternal()) {
 257                          logger.debug("Processing update of {}:{} node {} intf {} network {}",
 258                                  tableName, action, node, intfUUID, network.getNetworkUUID());
 259                         tenantNetworkManager.programInternalVlan(node, uuid, network);
 260                         this.handleInterfaceUpdate(node, intfUUID.toString(), intf);
 261                     } else {
 262                         logger.trace("Ignoring update because there is not a neutron network {} for port {}, interface {}",
 263                                 network, uuid, intfUUID);
 264                     }
 265                 } catch (Exception e) {
 266                     logger.error("Failed to process row update", e);
 267                 }
 268             }
 269         } else if (tableName.equalsIgnoreCase(ovsdbConfigurationService.getTableName(node, OpenVSwitch.class))) {
 270             logger.debug("Processing update of {}:{} node: {}, ovs uuid: {}, row: {}", tableName, action, node, uuid, row);
 271             try {
 272                 ConcurrentMap<String, Row> interfaces = this.ovsdbConfigurationService
 273                         .getRows(node, ovsdbConfigurationService.getTableName(node, Interface.class));
 274                 if (interfaces != null) {
 275                     for (String intfUUID : interfaces.keySet()) {
 276                         Interface intf = ovsdbConfigurationService.getTypedRow(node, Interface.class, interfaces.get(intfUUID));
 277                         this.handleInterfaceUpdate(node, intfUUID, intf);
 278                     }
 279                 }
 280             } catch (Exception e) {
 281                 logger.error("Error fetching Interface Rows for node " + node, e);
 282             }
 283         }
 284     }
 285
 286     private void handleInterfaceUpdate (Node node, String uuid, Interface intf) {
 287         logger.trace("Interface update of node: {}, uuid: {}", node, uuid);
 288         NeutronNetwork network = tenantNetworkManager.getTenantNetwork(intf);
 289         if (network != null) {
 290             if (bridgeConfigurationManager.createLocalNetwork(node, network))
 291                 networkingProviderManager.getProvider(node).handleInterfaceUpdate(network, node, intf);
 292         } else {
 293             logger.debug("No tenant network found on node: {}, uuid: {} for interface: {}", node, uuid, intf);
 294         }
 295     }
 296
 297     private void handleInterfaceDelete (Node node, String uuid, Interface intf, boolean isLastInstanceOnNode,
 298                                         NeutronNetwork network) {
 299         logger.debug("handleInterfaceDelete: node: {}, uuid: {}, isLastInstanceOnNode: {}, interface: {}",
 300                 node, uuid, isLastInstanceOnNode, intf);
 301
 302         List<String> phyIfName = bridgeConfigurationManager.getAllPhysicalInterfaceNames(node);
 303         if (intf.getTypeColumn().getData().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) ||
 304             intf.getTypeColumn().getData().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) ||
 305             phyIfName.contains(intf.getName())) {
 306             /* delete tunnel or physical interfaces */
 307             networkingProviderManager.getProvider(node).handleInterfaceDelete(intf.getTypeColumn().getData(), null, node, intf, isLastInstanceOnNode);
 308         } else if (network != null) {
 309             if (!network.getProviderNetworkType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VLAN)) { /* vlan doesn't need a tunnel endpoint */
 310                 if (configurationService.getTunnelEndPoint(node) == null) {
 311                     logger.error("Tunnel end-point configuration missing. Please configure it in OpenVSwitch Table");
 312                     return;
 313                 }
 314             }
 315             if (isLastInstanceOnNode & networkingProviderManager.getProvider(node).hasPerTenantTunneling()) {
 316                 tenantNetworkManager.reclaimInternalVlan(node, uuid, network);
 317             }
 318             networkingProviderManager.getProvider(node).handleInterfaceDelete(network.getProviderNetworkType(), network, node, intf, isLastInstanceOnNode);
 319         }
 320     }
 321
 322     private String getPortIdForInterface (Node node, String uuid, Interface intf) {
 323         try {
 324             Map<String, Row> ports = this.ovsdbConfigurationService.getRows(node, ovsdbConfigurationService.getTableName(node, Port.class));
 325             if (ports == null) return null;
 326             for (String portUUID : ports.keySet()) {
 327                 Port port = ovsdbConfigurationService.getTypedRow(node, Port.class, ports.get(portUUID));
 328                 Set<UUID> interfaceUUIDs = port.getInterfacesColumn().getData();
 329                 logger.trace("Scanning Port {} to identify interface : {} ",port, uuid);
 330                 for (UUID intfUUID : interfaceUUIDs) {
 331                     if (intfUUID.toString().equalsIgnoreCase(uuid)) {
 332                         logger.trace("Found Interface {} -> {}", uuid, portUUID);
 333                         return portUUID;
 334                     }
 335                 }
 336             }
 337         } catch (Exception e) {
 338             logger.debug("Failed to get Port tag for for Intf {}:{}", intf, e);
 339         }
 340         return null;
 341     }
 342
 343     @Override
 344     public void notifyNode(Node node, UpdateType type, Map<String, Property> propMap) {
 345         logger.debug("notifyNode: Node {} update {} from Controller's inventory Service", node, type);
 346
 347         // Add the Node Type check back once the Consistency issue is resolved between MD-SAL and AD-SAL
 348         if (!type.equals(UpdateType.REMOVED) && !nodeCache.contains(node)) {
 349             nodeCache.add(node);
 350             networkingProviderManager.getProvider(node).initializeOFFlowRules(node);
 351         } else if (type.equals(UpdateType.REMOVED)){
 352             nodeCache.remove(node);
 353         }
 354     }
 355
 356     @Override
 357     public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map<String, Property> propMap) {
 358         //We are not interested in the nodeConnectors at this moment
 359     }
 360
 361     private void triggerUpdates() {
 362         List<Node> nodes = connectionService.getNodes();
 363         if (nodes == null) return;
 364         for (Node node : nodes) {
 365             try {
 366                 List<String> tableNames = ovsdbConfigurationService.getTables(node);
 367                 if (tableNames == null) continue;
 368                 for (String tableName : tableNames) {
 369                     Map<String, Row> rows = ovsdbConfigurationService.getRows(node, tableName);
 370                     if (rows == null) continue;
 371                     for (String uuid : rows.keySet()) {
 372                         Row row = rows.get(uuid);
 373                         this.rowAdded(node, tableName, uuid, row);
 374                     }
 375                 }
 376             } catch (Exception e) {
 377                 logger.error("Exception during OVSDB Southbound update trigger", e);
 378             }
 379         }
 380     }
 381 }