neutron/src/main/java/org/opendaylight/ovsdb/neutron/SouthboundHandler.java

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