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