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