neutron/src/main/java/org/opendaylight/ovsdb/neutron/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.neutron;
  11
  12 import java.util.ArrayList;
  13 import java.util.List;
  14 import java.util.Map;
  15 import java.util.Set;
  16 import java.util.concurrent.BlockingQueue;
  17 import java.util.concurrent.ConcurrentMap;
  18 import java.util.concurrent.ExecutorService;
  19 import java.util.concurrent.Executors;
  20 import java.util.concurrent.LinkedBlockingQueue;
  21
  22 import org.opendaylight.controller.networkconfig.neutron.NeutronNetwork;
  23 import org.opendaylight.controller.sal.core.Node;
  24 import org.opendaylight.controller.sal.core.NodeConnector;
  25 import org.opendaylight.controller.sal.core.Property;
  26 import org.opendaylight.controller.sal.core.UpdateType;
  27 import org.opendaylight.controller.switchmanager.IInventoryListener;
  28 import org.opendaylight.ovsdb.lib.notation.UUID;
  29 import org.opendaylight.ovsdb.lib.table.Interface;
  30 import org.opendaylight.ovsdb.lib.table.Open_vSwitch;
  31 import org.opendaylight.ovsdb.lib.table.Port;
  32 import org.opendaylight.ovsdb.lib.table.internal.Table;
  33 import org.opendaylight.ovsdb.neutron.provider.IProviderNetworkManager;
  34 import org.opendaylight.ovsdb.plugin.OVSDBInventoryListener;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 public class SouthboundHandler extends BaseHandler implements OVSDBInventoryListener, IInventoryListener {
  39     static final Logger logger = LoggerFactory.getLogger(SouthboundHandler.class);
  40     //private Thread eventThread;
  41     private ExecutorService eventHandler;
  42     private BlockingQueue<SouthboundEvent> events;
  43     List<Node> nodeCache;
  44
  45     // The implementation for each of these services is resolved by the OSGi Service Manager
  46     private volatile IAdminConfigManager adminConfigManager;
  47     private volatile IInternalNetworkManager internalNetworkManager;
  48     private volatile ITenantNetworkManager tenantNetworkManager;
  49     private volatile IProviderNetworkManager providerNetworkManager;
  50
  51     void init() {
  52         eventHandler = Executors.newSingleThreadExecutor();
  53         this.events = new LinkedBlockingQueue<SouthboundEvent>();
  54         nodeCache = new ArrayList<>();
  55     }
  56
  57     void start() {
  58         eventHandler.submit(new Runnable()  {
  59             @Override
  60             public void run() {
  61                 while (true) {
  62                     SouthboundEvent ev;
  63                     try {
  64                         ev = events.take();
  65                     } catch (InterruptedException e) {
  66                         logger.info("The event handler thread was interrupted, shutting down", e);
  67                         return;
  68                     }
  69                     switch (ev.getType()) {
  70                     case NODE:
  71                         try {
  72                             processNodeUpdate(ev.getNode(), ev.getAction());
  73                         } catch (Exception e) {
  74                             logger.error("Exception caught in ProcessNodeUpdate for node " + ev.getNode(), e);
  75                         }
  76                         break;
  77                     case ROW:
  78                         try {
  79                             processRowUpdate(ev.getNode(), ev.getTableName(), ev.getUuid(), ev.getRow(),
  80                                              ev.getContext(),ev.getAction());
  81                         } catch (Exception e) {
  82                             logger.error("Exception caught in ProcessRowUpdate for node " + ev.getNode(), e);
  83                         }
  84                         break;
  85                     default:
  86                         logger.warn("Unable to process action " + ev.getAction() + " for node " + ev.getNode());
  87                     }
  88                 }
  89             }
  90         });
  91         this.triggerUpdates();
  92     }
  93
  94     void stop() {
  95         eventHandler.shutdownNow();
  96     }
  97
  98     @Override
  99     public void nodeAdded(Node node) {
 100         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.ADD));
 101     }
 102
 103     @Override
 104     public void nodeRemoved(Node node) {
 105         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.DELETE));
 106     }
 107
 108     @Override
 109     public void rowAdded(Node node, String tableName, String uuid, Table<?> row) {
 110         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.ADD));
 111     }
 112
 113     @Override
 114     public void rowUpdated(Node node, String tableName, String uuid, Table<?> oldRow, Table<?> newRow) {
 115         if (this.isUpdateOfInterest(node, oldRow, newRow)) {
 116             this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, newRow, SouthboundEvent.Action.UPDATE));
 117         }
 118     }
 119
 120     /*
 121      * Ignore unneccesary updates to be even considered for processing.
 122      * (Especially stats update are fast and furious).
 123      */
 124
 125     private boolean isUpdateOfInterest(Node node, Table<?> oldRow, Table<?> newRow) {
 126         if (oldRow == null) return true;
 127         if (newRow.getTableName().equals(Interface.NAME)) {
 128             // We are NOT interested in Stats only updates
 129             Interface oldIntf = (Interface)oldRow;
 130             if (oldIntf.getName() == null && oldIntf.getExternal_ids() == null && oldIntf.getMac() == null &&
 131                 oldIntf.getOfport() == null && oldIntf.getOptions() == null && oldIntf.getOther_config() == null &&
 132                 oldIntf.getType() == null) {
 133                 logger.trace("IGNORING Interface Update: node {}, row: {}", node, newRow);
 134                 return false;
 135             }
 136         } else if (newRow.getTableName().equals(Port.NAME)) {
 137             // We are NOT interested in Stats only updates
 138             Port oldPort = (Port)oldRow;
 139             if (oldPort.getName() == null && oldPort.getExternal_ids() == null && oldPort.getMac() == null &&
 140                 oldPort.getInterfaces() == null && oldPort.getTag() == null && oldPort.getTrunks() == null) {
 141                 logger.trace("IGNORING Port Update: node {}, row: {}", node, newRow);
 142                 return false;
 143             }
 144         } else if (newRow.getTableName().equals(Open_vSwitch.NAME)) {
 145             Open_vSwitch oldOpenvSwitch = (Open_vSwitch) oldRow;
 146             if (oldOpenvSwitch.getOther_config()== null) {
 147                 /* we are only interested in other_config field change */
 148                 return false;
 149             }
 150         }
 151         return true;
 152     }
 153
 154     @Override
 155     public void rowRemoved(Node node, String tableName, String uuid, Table<?> row, Object context) {
 156         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, context, SouthboundEvent.Action.DELETE));
 157     }
 158
 159     private void enqueueEvent (SouthboundEvent event) {
 160         try {
 161             events.put(event);
 162         } catch (InterruptedException e) {
 163             logger.error("Thread was interrupted while trying to enqueue event ", e);
 164         }
 165     }
 166
 167     public void processNodeUpdate(Node node, SouthboundEvent.Action action) {
 168         if (action == SouthboundEvent.Action.DELETE) return;
 169         logger.trace("Process Node added {}", node);
 170         internalNetworkManager.prepareInternalNetwork(node);
 171     }
 172
 173     private void processRowUpdate(Node node, String tableName, String uuid, Table<?> row,
 174                                   Object context, SouthboundEvent.Action action) {
 175         if (action == SouthboundEvent.Action.DELETE) {
 176             if (Interface.NAME.getName().equalsIgnoreCase(tableName)) {
 177                 logger.debug("processRowUpdate: {} Deleted node: {}, uuid: {}, row: {}", tableName, node, uuid, row);
 178                 Interface deletedIntf = (Interface)row;
 179                 NeutronNetwork network = null;
 180                 if (context == null) {
 181                     network = tenantNetworkManager.getTenantNetworkForInterface(deletedIntf);
 182                 } else {
 183                     network = (NeutronNetwork)context;
 184                 }
 185                 List<String> phyIfName = adminConfigManager.getAllPhysicalInterfaceNames(node);
 186                 logger.info("Delete interface " + deletedIntf.getName());
 187
 188                 if (deletedIntf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) ||
 189                     deletedIntf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) ||
 190                     phyIfName.contains(deletedIntf.getName())) {
 191                     /* delete tunnel interfaces or physical interfaces */
 192                     this.handleInterfaceDelete(node, uuid, deletedIntf, false, null);
 193                 } else if (network != null && !network.getRouterExternal()) {
 194                     try {
 195                         ConcurrentMap<String, Table<?>> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName());
 196                         if (interfaces != null) {
 197                             boolean isLastInstanceOnNode = true;
 198                             for (String intfUUID : interfaces.keySet()) {
 199                                 if (intfUUID.equals(uuid)) continue;
 200                                 Interface intf = (Interface) interfaces.get(intfUUID);
 201                                 NeutronNetwork neutronNetwork = tenantNetworkManager.getTenantNetworkForInterface(intf);
 202                                 if (neutronNetwork != null && neutronNetwork.equals(network)) isLastInstanceOnNode = false;
 203                             }
 204                             this.handleInterfaceDelete(node, uuid, deletedIntf, isLastInstanceOnNode, network);
 205                         }
 206                     } catch (Exception e) {
 207                         logger.error("Error fetching Interface Rows for node " + node, e);
 208                     }
 209                 }
 210             }
 211         }
 212         else if (Interface.NAME.getName().equalsIgnoreCase(tableName)) {
 213             logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row);
 214             Interface intf = (Interface)row;
 215             NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf);
 216             if (network != null && !network.getRouterExternal()) {
 217                 if (providerNetworkManager.getProvider().hasPerTenantTunneling()) {
 218                     int vlan = tenantNetworkManager.networkCreated(node, network.getID());
 219                     logger.trace("Neutron Network {}:{} Created with Internal Vlan: {}", network.getNetworkUUID(), network.getNetworkName(), vlan);
 220
 221                     String portUUID = this.getPortIdForInterface(node, uuid, intf);
 222                     if (portUUID != null) {
 223                         tenantNetworkManager.programTenantNetworkInternalVlan(node, portUUID, network);
 224                     }
 225                 }
 226                 this.handleInterfaceUpdate(node, uuid, intf);
 227             }
 228         } else if (Port.NAME.getName().equalsIgnoreCase(tableName)) {
 229             logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row);
 230             Port port = (Port)row;
 231             Set<UUID> interfaceUUIDs = port.getInterfaces();
 232             for (UUID intfUUID : interfaceUUIDs) {
 233                 logger.trace("Scanning interface "+intfUUID);
 234                 try {
 235                     Interface intf = (Interface)this.ovsdbConfigService.getRow(node, Interface.NAME.getName(), intfUUID.toString());
 236                     NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf);
 237                     if (network != null && !network.getRouterExternal()) {
 238                         tenantNetworkManager.programTenantNetworkInternalVlan(node, uuid, network);
 239                     } else {
 240                         logger.trace("ignore update because there is not a neutron network.");
 241                     }
 242                 } catch (Exception e) {
 243                     logger.error("Failed to process row update", e);
 244                 }
 245             }
 246         } else if (Open_vSwitch.NAME.getName().equalsIgnoreCase(tableName)) {
 247             logger.debug("processRowUpdate: {} Added / Updated node: {}, uuid: {}, row: {}", tableName, node, uuid, row);
 248             try {
 249                 ConcurrentMap<String, Table<?>> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName());
 250                 if (interfaces != null) {
 251                     for (String intfUUID : interfaces.keySet()) {
 252                         Interface intf = (Interface) interfaces.get(intfUUID);
 253                         this.handleInterfaceUpdate(node, intfUUID, intf);
 254                     }
 255                 }
 256             } catch (Exception e) {
 257                 logger.error("Error fetching Interface Rows for node " + node, e);
 258             }
 259         }
 260     }
 261
 262     private void handleInterfaceUpdate (Node node, String uuid, Interface intf) {
 263         logger.debug("handleInterfaceUpdate: node: {}, uuid: {}", node, uuid);
 264         NeutronNetwork network = tenantNetworkManager.getTenantNetworkForInterface(intf);
 265         if (network != null) {
 266             if (internalNetworkManager.checkAndCreateNetwork(node, network)) {
 267                 providerNetworkManager.getProvider().handleInterfaceUpdate(network, node, intf);
 268             }
 269         }
 270     }
 271
 272     private void handleInterfaceDelete (Node node, String uuid, Interface intf, boolean isLastInstanceOnNode,
 273                                         NeutronNetwork network) {
 274         logger.debug("handleInterfaceDelete: node: {}, uuid: {}, isLastInstanceOnNode: {}, interface: {}",
 275                 node, uuid, isLastInstanceOnNode, intf);
 276
 277         List<String> phyIfName = adminConfigManager.getAllPhysicalInterfaceNames(node);
 278         if (intf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VXLAN) ||
 279             intf.getType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_GRE) ||
 280             phyIfName.contains(intf.getName())) {
 281             /* delete tunnel or physical interfaces */
 282             providerNetworkManager.getProvider().handleInterfaceDelete(intf.getType(), null, node, intf, isLastInstanceOnNode);
 283         } else if (network != null) {
 284             if (!network.getProviderNetworkType().equalsIgnoreCase(NetworkHandler.NETWORK_TYPE_VLAN)) { /* vlan doesn't need a tunnel endpoint */
 285                 if (adminConfigManager.getTunnelEndPoint(node) == null) {
 286                     logger.error("Tunnel end-point configuration missing. Please configure it in Open_vSwitch Table");
 287                     return;
 288                 }
 289             }
 290             if (isLastInstanceOnNode & providerNetworkManager.getProvider().hasPerTenantTunneling()) {
 291                 tenantNetworkManager.reclaimTenantNetworkInternalVlan(node, uuid, network);
 292             }
 293             providerNetworkManager.getProvider().handleInterfaceDelete(network.getProviderNetworkType(), network, node, intf, isLastInstanceOnNode);
 294         }
 295     }
 296
 297     private String getPortIdForInterface (Node node, String uuid, Interface intf) {
 298         try {
 299             Map<String, Table<?>> ports = this.ovsdbConfigService.getRows(node, Port.NAME.getName());
 300             if (ports == null) return null;
 301             for (String portUUID : ports.keySet()) {
 302                 Port port = (Port)ports.get(portUUID);
 303                 Set<UUID> interfaceUUIDs = port.getInterfaces();
 304                 logger.trace("Scanning Port {} to identify interface : {} ",port, uuid);
 305                 for (UUID intfUUID : interfaceUUIDs) {
 306                     if (intfUUID.toString().equalsIgnoreCase(uuid)) {
 307                         logger.trace("Found Interface {} -> {}", uuid, portUUID);
 308                         return portUUID;
 309                     }
 310                 }
 311             }
 312         } catch (Exception e) {
 313             logger.debug("Failed to add Port tag for for Intf {}",intf, e);
 314         }
 315         return null;
 316     }
 317
 318     @Override
 319     public void notifyNode(Node node, UpdateType type, Map<String, Property> propMap) {
 320         logger.debug("notifyNode: Node {} update {} from Controller's inventory Service", node, type);
 321
 322         // Add the Node Type check back once the Consistency issue is resolved between MD-SAL and AD-SAL
 323         if (!type.equals(UpdateType.REMOVED) && !nodeCache.contains(node)) {
 324             nodeCache.add(node);
 325             providerNetworkManager.getProvider().initializeOFFlowRules(node);
 326         } else if (type.equals(UpdateType.REMOVED)){
 327             nodeCache.remove(node);
 328         }
 329     }
 330
 331     @Override
 332     public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map<String, Property> propMap) {
 333         //We are not interested in the nodeConnectors at this moment
 334     }
 335
 336     private void triggerUpdates() {
 337         List<Node> nodes = this.getConnectionService().getNodes();
 338         if (nodes == null) return;
 339         for (Node node : nodes) {
 340             try {
 341                 List<String> tableNames = this.getOVSDBConfigService().getTables(node);
 342                 if (tableNames == null) continue;
 343                 for (String tableName : tableNames) {
 344                     Map<String, Table<?>> rows = this.getOVSDBConfigService().getRows(node, tableName);
 345                     if (rows == null) continue;
 346                     for (String uuid : rows.keySet()) {
 347                         Table<?> row = rows.get(uuid);
 348                         this.rowAdded(node, tableName, uuid, row);
 349                     }
 350                 }
 351             } catch (Exception e) {
 352                 logger.error("Exception during OVSDB Southbound update trigger", e);
 353             }
 354         }
 355     }
 356 }