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
   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.Table;
  33 import org.opendaylight.ovsdb.neutron.provider.ProviderNetworkManager;
  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     void init() {
  46         eventHandler = Executors.newSingleThreadExecutor();
  47         this.events = new LinkedBlockingQueue<SouthboundEvent>();
  48         nodeCache = new ArrayList<>();
  49     }
  50
  51     void start() {
  52         eventHandler.submit(new Runnable()  {
  53             @Override
  54             public void run() {
  55                 while (true) {
  56                     SouthboundEvent ev;
  57                     try {
  58                         ev = events.take();
  59                     } catch (InterruptedException e) {
  60                         logger.info("The event handler thread was interrupted, shutting down", e);
  61                         return;
  62                     }
  63                     switch (ev.getType()) {
  64                     case NODE:
  65                         try {
  66                             processNodeUpdate(ev.getNode(), ev.getAction());
  67                         } catch (Exception e) {
  68                             logger.error("Exception caught in ProcessNodeUpdate for node " + ev.getNode(), e);
  69                         }
  70                         break;
  71                     case ROW:
  72                         try {
  73                             processRowUpdate(ev.getNode(), ev.getTableName(), ev.getUuid(), ev.getRow(), ev.getAction());
  74                         } catch (Exception e) {
  75                             logger.error("Exception caught in ProcessRowUpdate for node " + ev.getNode(), e);
  76                         }
  77                         break;
  78                     default:
  79                         logger.warn("Unable to process action " + ev.getAction() + " for node " + ev.getNode());
  80                     }
  81                 }
  82             }
  83         });
  84         this.triggerUpdates();
  85     }
  86
  87     void stop() {
  88         eventHandler.shutdownNow();
  89     }
  90
  91     @Override
  92     public void nodeAdded(Node node) {
  93         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.ADD));
  94     }
  95
  96     @Override
  97     public void nodeRemoved(Node node) {
  98         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.DELETE));
  99     }
 100
 101     @Override
 102     public void rowAdded(Node node, String tableName, String uuid, Table<?> row) {
 103         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.ADD));
 104     }
 105
 106     @Override
 107     public void rowUpdated(Node node, String tableName, String uuid, Table<?> oldRow, Table<?> newRow) {
 108         if (this.isUpdateOfInterest(oldRow, newRow)) {
 109             this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, newRow, SouthboundEvent.Action.UPDATE));
 110         }
 111     }
 112
 113     /*
 114      * Ignore unneccesary updates to be even considered for processing.
 115      * (Especially stats update are fast and furious).
 116      */
 117
 118     private boolean isUpdateOfInterest(Table<?> oldRow, Table<?> newRow) {
 119         if (oldRow == null) return true;
 120         if (newRow.getTableName().equals(Interface.NAME)) {
 121             // We are NOT interested in Stats only updates
 122             Interface oldIntf = (Interface)oldRow;
 123             if (oldIntf.getName() == null && oldIntf.getExternal_ids() == null && oldIntf.getMac() == null &&
 124                 oldIntf.getOfport() == null && oldIntf.getOptions() == null && oldIntf.getOther_config() == null &&
 125                 oldIntf.getType() == null) {
 126                 logger.trace("IGNORING Interface Update : "+newRow.toString());
 127                 return false;
 128             }
 129         } else if (newRow.getTableName().equals(Port.NAME)) {
 130             // We are NOT interested in Stats only updates
 131             Port oldPort = (Port)oldRow;
 132             if (oldPort.getName() == null && oldPort.getExternal_ids() == null && oldPort.getMac() == null &&
 133                 oldPort.getInterfaces() == null && oldPort.getTag() == null && oldPort.getTrunks() == null) {
 134                 logger.trace("IGNORING Port Update : "+newRow.toString());
 135                 return false;
 136             }
 137         }
 138
 139         return true;
 140     }
 141
 142     @Override
 143     public void rowRemoved(Node node, String tableName, String uuid, Table<?> row) {
 144         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.DELETE));
 145     }
 146
 147     private void enqueueEvent (SouthboundEvent event) {
 148         try {
 149             events.put(event);
 150         } catch (InterruptedException e) {
 151             logger.error("Thread was interrupted while trying to enqueue event ", e);
 152         }
 153
 154     }
 155
 156     public void processNodeUpdate(Node node, SouthboundEvent.Action action) {
 157         if (action == SouthboundEvent.Action.DELETE) return;
 158         logger.trace("Process Node added {}", node);
 159         InternalNetworkManager.getManager().prepareInternalNetwork(node);
 160     }
 161
 162     private void processRowUpdate(Node node, String tableName, String uuid, Table<?> row,
 163                                   SouthboundEvent.Action action) {
 164         if (action == SouthboundEvent.Action.DELETE) {
 165             if (Interface.NAME.getName().equalsIgnoreCase(tableName)) {
 166                 Interface deletedIntf = (Interface)row;
 167                 NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(deletedIntf);
 168                 if (network != null && !network.getRouterExternal()) {
 169                     try {
 170                         ConcurrentMap<String, Table<?>> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName());
 171                         if (interfaces != null) {
 172                             boolean isLastInstanceOnNode = true;
 173                             for (String intfUUID : interfaces.keySet()) {
 174                                 if (intfUUID.equals(uuid)) continue;
 175                                 Interface intf = (Interface) interfaces.get(intfUUID);
 176                                 NeutronNetwork neutronNetwork = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 177                                 if (neutronNetwork != null && neutronNetwork.equals(network)) isLastInstanceOnNode = false;
 178                             }
 179                             this.handleInterfaceDelete(node, uuid, deletedIntf, isLastInstanceOnNode);
 180                         }
 181                     } catch (Exception e) {
 182                         logger.error("Error fetching Interface Rows for node " + node, e);
 183                     }
 184                 }
 185             }
 186         }
 187         else if (Interface.NAME.getName().equalsIgnoreCase(tableName)) {
 188             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 189             Interface intf = (Interface)row;
 190             NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 191             if (network != null && !network.getRouterExternal()) {
 192                 if (ProviderNetworkManager.getManager().hasPerTenantTunneling()) {
 193                     int vlan = TenantNetworkManager.getManager().networkCreated(node, network.getID());
 194                     logger.trace("Neutron Network {} Created with Internal Vlan : {}", network.toString(), vlan);
 195
 196                     String portUUID = this.getPortIdForInterface(node, uuid, intf);
 197                     if (portUUID != null) {
 198                         TenantNetworkManager.getManager().programTenantNetworkInternalVlan(node, portUUID, network);
 199                     }
 200                 }
 201                 this.handleInterfaceUpdate(node, uuid, intf);
 202             }
 203         } else if (Port.NAME.getName().equalsIgnoreCase(tableName)) {
 204             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 205             Port port = (Port)row;
 206             Set<UUID> interfaceUUIDs = port.getInterfaces();
 207             for (UUID intfUUID : interfaceUUIDs) {
 208                 logger.trace("Scanning interface "+intfUUID);
 209                 try {
 210                     Interface intf = (Interface)this.ovsdbConfigService.getRow(node, Interface.NAME.getName(), intfUUID.toString());
 211                     NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 212                     if (network != null && !network.getRouterExternal()) {
 213                         TenantNetworkManager.getManager().programTenantNetworkInternalVlan(node, uuid, network);
 214                     }
 215                 } catch (Exception e) {
 216                     logger.error("Failed to process row update", e);
 217                 }
 218             }
 219         } else if (Open_vSwitch.NAME.getName().equalsIgnoreCase(tableName)) {
 220             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 221             try {
 222                 ConcurrentMap<String, Table<?>> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName());
 223                 if (interfaces != null) {
 224                     for (String intfUUID : interfaces.keySet()) {
 225                         Interface intf = (Interface) interfaces.get(intfUUID);
 226                         this.handleInterfaceUpdate(node, intfUUID, intf);
 227                     }
 228                 }
 229             } catch (Exception e) {
 230                 logger.error("Error fetching Interface Rows for node " + node, e);
 231             }
 232         }
 233     }
 234
 235     private void handleInterfaceUpdate (Node node, String uuid, Interface intf) {
 236         if (AdminConfigManager.getManager().getTunnelEndPoint(node) == null) {
 237             logger.error("Tunnel end-point configuration missing. Please configure it in Open_vSwitch Table");
 238             return;
 239         }
 240         NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 241         if (network != null) {
 242             ProviderNetworkManager.getManager().handleInterfaceUpdate(network.getProviderNetworkType(),
 243                     network.getProviderSegmentationID(), node, intf);
 244         }
 245     }
 246     private void handleInterfaceDelete (Node node, String uuid, Interface intf, boolean isLastInstanceOnNode) {
 247         if (AdminConfigManager.getManager().getTunnelEndPoint(node) == null) {
 248             logger.error("Tunnel end-point configuration missing. Please configure it in Open_vSwitch Table");
 249             return;
 250         }
 251         NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 252         if (network != null) {
 253             if (isLastInstanceOnNode) {
 254                 TenantNetworkManager.getManager().reclaimTennantNetworkInternalVlan(node, uuid, network);
 255             }
 256             ProviderNetworkManager.getManager().handleInterfaceDelete(network.getProviderNetworkType(),
 257                     network.getProviderSegmentationID(), node, intf, isLastInstanceOnNode);
 258         }
 259     }
 260
 261     private String getPortIdForInterface (Node node, String uuid, Interface intf) {
 262         try {
 263             Map<String, Table<?>> ports = this.ovsdbConfigService.getRows(node, Port.NAME.getName());
 264             if (ports == null) return null;
 265             for (String portUUID : ports.keySet()) {
 266                 Port port = (Port)ports.get(portUUID);
 267                 Set<UUID> interfaceUUIDs = port.getInterfaces();
 268                 logger.trace("Scanning Port {} to identify interface : {} ",port, uuid);
 269                 for (UUID intfUUID : interfaceUUIDs) {
 270                     if (intfUUID.toString().equalsIgnoreCase(uuid)) {
 271                         logger.trace("Found Interafce {} -> {}", uuid, portUUID);
 272                         return portUUID;
 273                     }
 274                 }
 275             }
 276         } catch (Exception e) {
 277             logger.debug("Failed to add Port tag for for Intf {}",intf, e);
 278         }
 279         return null;
 280     }
 281
 282     @Override
 283     public void notifyNode(Node node, UpdateType type, Map<String, Property> propMap) {
 284         logger.debug("Node {} update {} from Controller's inventory Service", node, type);
 285
 286         // Add the Node Type check back once the Consistency issue is resolved between MD-SAL and AD-SAL
 287         if (!type.equals(UpdateType.REMOVED) && !nodeCache.contains(node)) {
 288             nodeCache.add(node);
 289             ProviderNetworkManager.getManager().initializeOFFlowRules(node);
 290         } else if (type.equals(UpdateType.REMOVED)){
 291             nodeCache.remove(node);
 292         }
 293     }
 294
 295     @Override
 296     public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map<String, Property> propMap) {
 297         //We are not interested in the nodeConnectors at this moment
 298     }
 299
 300     private void triggerUpdates() {
 301         List<Node> nodes = this.getConnectionService().getNodes();
 302         if (nodes == null) return;
 303         for (Node node : nodes) {
 304             try {
 305                 List<String> tableNames = this.getOVSDBConfigService().getTables(node);
 306                 if (tableNames == null) continue;
 307                 for (String tableName : tableNames) {
 308                     Map<String, Table<?>> rows = this.getOVSDBConfigService().getRows(node, tableName);
 309                     if (rows == null) continue;
 310                     for (String uuid : rows.keySet()) {
 311                         Table<?> row = rows.get(uuid);
 312                         this.rowAdded(node, tableName, uuid, row);
 313                     }
 314                 }
 315             } catch (Exception e) {
 316                 logger.error("Exception during OVSDB Southbound update trigger", e);
 317             }
 318         }
 319     }
 320 }