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