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.List;
  13 import java.util.Map;
  14 import java.util.Set;
  15 import java.util.concurrent.BlockingQueue;
  16 import java.util.concurrent.ExecutorService;
  17 import java.util.concurrent.Executors;
  18 import java.util.concurrent.LinkedBlockingQueue;
  19
  20 import org.opendaylight.controller.networkconfig.neutron.NeutronNetwork;
  21 import org.opendaylight.controller.sal.core.Node;
  22 import org.opendaylight.controller.sal.core.NodeConnector;
  23 import org.opendaylight.controller.sal.core.Property;
  24 import org.opendaylight.controller.sal.core.UpdateType;
  25 import org.opendaylight.controller.switchmanager.IInventoryListener;
  26 import org.opendaylight.ovsdb.lib.notation.UUID;
  27 import org.opendaylight.ovsdb.lib.table.Interface;
  28 import org.opendaylight.ovsdb.lib.table.Open_vSwitch;
  29 import org.opendaylight.ovsdb.lib.table.Port;
  30 import org.opendaylight.ovsdb.lib.table.internal.Table;
  31 import org.opendaylight.ovsdb.neutron.provider.ProviderNetworkManager;
  32 import org.opendaylight.ovsdb.plugin.OVSDBInventoryListener;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 public class SouthboundHandler extends BaseHandler implements OVSDBInventoryListener, IInventoryListener {
  37     static final Logger logger = LoggerFactory.getLogger(SouthboundHandler.class);
  38     //private Thread eventThread;
  39     private ExecutorService eventHandler;
  40     private BlockingQueue<SouthboundEvent> events;
  41
  42     void init() {
  43         eventHandler = Executors.newSingleThreadExecutor();
  44         this.events = new LinkedBlockingQueue<SouthboundEvent>();
  45     }
  46
  47     void start() {
  48         eventHandler.submit(new Runnable()  {
  49             @Override
  50             public void run() {
  51                 while (true) {
  52                     SouthboundEvent ev;
  53                     try {
  54                         ev = events.take();
  55                     } catch (InterruptedException e) {
  56                         logger.info("The event handler thread was interrupted, shutting down", e);
  57                         return;
  58                     }
  59                     switch (ev.getType()) {
  60                     case NODE:
  61                         try {
  62                             processNodeUpdate(ev.getNode(), ev.getAction());
  63                         } catch (Exception e) {
  64                             logger.error("Exception caught in ProcessNodeUpdate for node " + ev.getNode(), e);
  65                         }
  66                         break;
  67                     case ROW:
  68                         try {
  69                             processRowUpdate(ev.getNode(), ev.getTableName(), ev.getUuid(), ev.getRow(), ev.getAction());
  70                         } catch (Exception e) {
  71                             logger.error("Exception caught in ProcessRowUpdate for node " + ev.getNode(), e);
  72                         }
  73                         break;
  74                     default:
  75                         logger.warn("Unable to process action " + ev.getAction() + " for node " + ev.getNode());
  76                     }
  77                 }
  78             }
  79         });
  80         this.triggerUpdates();
  81     }
  82
  83     void stop() {
  84         eventHandler.shutdownNow();
  85     }
  86
  87     @Override
  88     public void nodeAdded(Node node) {
  89         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.ADD));
  90     }
  91
  92     @Override
  93     public void nodeRemoved(Node node) {
  94         this.enqueueEvent(new SouthboundEvent(node, SouthboundEvent.Action.DELETE));
  95     }
  96
  97     @Override
  98     public void rowAdded(Node node, String tableName, String uuid, Table<?> row) {
  99         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.ADD));
 100     }
 101
 102     @Override
 103     public void rowUpdated(Node node, String tableName, String uuid, Table<?> oldRow, Table<?> newRow) {
 104         if (this.isUpdateOfInterest(oldRow, newRow)) {
 105             this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, newRow, SouthboundEvent.Action.UPDATE));
 106         }
 107     }
 108
 109     /*
 110      * Ignore unneccesary updates to be even considered for processing.
 111      * (Especially stats update are fast and furious).
 112      */
 113
 114     private boolean isUpdateOfInterest(Table<?> oldRow, Table<?> newRow) {
 115         if (oldRow == null) return true;
 116         if (newRow.getTableName().equals(Interface.NAME)) {
 117             // We are NOT interested in Stats only updates
 118             Interface oldIntf = (Interface)oldRow;
 119             if (oldIntf.getName() == null && oldIntf.getExternal_ids() == null && oldIntf.getMac() == null &&
 120                 oldIntf.getOfport() == null && oldIntf.getOptions() == null && oldIntf.getOther_config() == null &&
 121                 oldIntf.getType() == null) {
 122                 logger.trace("IGNORING Interface Update : "+newRow.toString());
 123                 return false;
 124             }
 125         } else if (newRow.getTableName().equals(Port.NAME)) {
 126             // We are NOT interested in Stats only updates
 127             Port oldPort = (Port)oldRow;
 128             if (oldPort.getName() == null && oldPort.getExternal_ids() == null && oldPort.getMac() == null &&
 129                 oldPort.getInterfaces() == null && oldPort.getTag() == null && oldPort.getTrunks() == null) {
 130                 logger.trace("IGNORING Port Update : "+newRow.toString());
 131                 return false;
 132             }
 133         }
 134
 135         return true;
 136     }
 137
 138     @Override
 139     public void rowRemoved(Node node, String tableName, String uuid, Table<?> row) {
 140         this.enqueueEvent(new SouthboundEvent(node, tableName, uuid, row, SouthboundEvent.Action.DELETE));
 141     }
 142
 143     private void enqueueEvent (SouthboundEvent event) {
 144         try {
 145             events.put(event);
 146         } catch (InterruptedException e) {
 147             logger.error("Thread was interrupted while trying to enqueue event ", e);
 148         }
 149
 150     }
 151
 152     public void processNodeUpdate(Node node, SouthboundEvent.Action action) {
 153         if (action == SouthboundEvent.Action.DELETE) return;
 154         logger.trace("Process Node added {}", node);
 155         InternalNetworkManager.getManager().prepareInternalNetwork(node);
 156     }
 157
 158     private void processRowUpdate(Node node, String tableName, String uuid, Table<?> row,
 159                                   SouthboundEvent.Action action) {
 160         if (action == SouthboundEvent.Action.DELETE) return;
 161
 162         if (Interface.NAME.getName().equalsIgnoreCase(tableName)) {
 163             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 164             Interface intf = (Interface)row;
 165             NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 166             if (network != null && !network.getRouterExternal()) {
 167                 if (ProviderNetworkManager.getManager().hasPerTenantTunneling()) {
 168                     int vlan = TenantNetworkManager.getManager().networkCreated(network.getID());
 169                     logger.trace("Neutron Network {} Created with Internal Vlan : {}", network.toString(), vlan);
 170
 171                     String portUUID = this.getPortIdForInterface(node, uuid, intf);
 172                     if (portUUID != null) {
 173                         TenantNetworkManager.getManager().programTenantNetworkInternalVlan(node, portUUID, network);
 174                     }
 175                 }
 176                 this.handleInterfaceUpdate(node, uuid, intf);
 177             }
 178         } else if (Port.NAME.getName().equalsIgnoreCase(tableName)) {
 179             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 180             Port port = (Port)row;
 181             Set<UUID> interfaceUUIDs = port.getInterfaces();
 182             for (UUID intfUUID : interfaceUUIDs) {
 183                 logger.trace("Scanning interface "+intfUUID);
 184                 try {
 185                     Interface intf = (Interface)this.ovsdbConfigService.getRow(node, Interface.NAME.getName(), intfUUID.toString());
 186                     NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 187                     if (network != null && !network.getRouterExternal()) {
 188                         TenantNetworkManager.getManager().programTenantNetworkInternalVlan(node, uuid, network);
 189                     }
 190                 } catch (Exception e) {
 191                     logger.error("Failed to process row update", e);
 192                 }
 193             }
 194         } else if (Open_vSwitch.NAME.getName().equalsIgnoreCase(tableName)) {
 195             logger.debug("{} Added / Updated {} , {}, {}", tableName, node, uuid, row);
 196             try {
 197                 Map<String, Table<?>> interfaces = this.ovsdbConfigService.getRows(node, Interface.NAME.getName());
 198                 if (interfaces != null) {
 199                     for (String intfUUID : interfaces.keySet()) {
 200                         Interface intf = (Interface) interfaces.get(intfUUID);
 201                         this.handleInterfaceUpdate(node, intfUUID, intf);
 202                     }
 203                 }
 204             } catch (Exception e) {
 205                 logger.error("Error fetching Interface Rows for node " + node, e);
 206             }
 207         }
 208     }
 209
 210     private void handleInterfaceUpdate (Node node, String uuid, Interface intf) {
 211         if (AdminConfigManager.getManager().getTunnelEndPoint(node) == null) {
 212             logger.error("Tunnel end-point configuration missing. Please configure it in Open_vSwitch Table");
 213             return;
 214         }
 215         NeutronNetwork network = TenantNetworkManager.getManager().getTenantNetworkForInterface(intf);
 216         if (network != null) {
 217             ProviderNetworkManager.getManager().handleInterfaceUpdate(network.getProviderNetworkType(),
 218                     network.getProviderSegmentationID(), node, intf);
 219         }
 220     }
 221
 222     private String getPortIdForInterface (Node node, String uuid, Interface intf) {
 223         try {
 224             Map<String, Table<?>> ports = this.ovsdbConfigService.getRows(node, Port.NAME.getName());
 225             if (ports == null) return null;
 226             for (String portUUID : ports.keySet()) {
 227                 Port port = (Port)ports.get(portUUID);
 228                 Set<UUID> interfaceUUIDs = port.getInterfaces();
 229                 logger.trace("Scanning Port {} to identify interface : {} ",port, uuid);
 230                 for (UUID intfUUID : interfaceUUIDs) {
 231                     if (intfUUID.toString().equalsIgnoreCase(uuid)) {
 232                         logger.trace("Found Interafce {} -> {}", uuid, portUUID);
 233                         return portUUID;
 234                     }
 235                 }
 236             }
 237         } catch (Exception e) {
 238             logger.debug("Failed to add Port tag for for Intf {}",intf, e);
 239         }
 240         return null;
 241     }
 242
 243     @Override
 244     public void notifyNode(Node node, UpdateType type, Map<String, Property> propMap) {
 245         if (node.getType().equals(Node.NodeIDType.OPENFLOW) && type.equals(UpdateType.ADDED)) {
 246             logger.debug("OpenFlow node {} added. Initialize Basic flows", node);
 247             ProviderNetworkManager.getManager().initializeOFFlowRules(node);
 248         }
 249     }
 250
 251     @Override
 252     public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map<String, Property> propMap) {
 253         //We are not interested in the nodeConnectors at this moment
 254     }
 255
 256     private void triggerUpdates() {
 257         List<Node> nodes = this.getConnectionService().getNodes();
 258         if (nodes == null) return;
 259         for (Node node : nodes) {
 260             try {
 261                 List<String> tableNames = this.getOVSDBConfigService().getTables(node);
 262                 if (tableNames == null) continue;
 263                 for (String tableName : tableNames) {
 264                     Map<String, Table<?>> rows = this.getOVSDBConfigService().getRows(node, tableName);
 265                     if (rows == null) continue;
 266                     for (String uuid : rows.keySet()) {
 267                         Table<?> row = rows.get(uuid);
 268                         this.rowAdded(node, tableName, uuid, row);
 269                     }
 270                 }
 271             } catch (Exception e) {
 272                 logger.error("Exception during OVSDB Southbound update trigger", e);
 273             }
 274         }
 275     }
 276 }