OVS plugin: add tunnel ips to central database

bp simplify-ovs-tunnel-mgmt

Makes the OVS plugin agent report its own IP address to the centralized
quantum database, and build its set of tunnels based on the contents of
this centralized database.  This removes the need for a removes need for
a 'remote-ips' file on each compute node that needs to be updated
when a new host is added.

Also:
- simplifies error handling within tunnel manager daemon_loop
- fixes issues with operational status not working for tunnel-mode
- fixes issue that not-stripping vlan of tunneled packet potentially
crashes OVS, causing flows to get wiped.

Change-Id: I4d285669e29beecf745fe620581fa6bc332a446c

etc/quantum/plugins/openvswitch/ovs_quantum_plugin.ini

@@ -20,15 +20,9 @@ enable-tunneling = False
 # connectivity.
 integration-bridge = br-int
 
-# Uncomment this line if enable-tunneling is True above.
+# Only used if enable-tunneling (above) is True.
 # In most cases, the default value should be fine.
-# tunnel-bridge = br-tun
+tunnel-bridge = br-tun
-
-# Uncomment this line if enable-tunneling is True above.
-# This file contains a list of IP addresses (one per line) that point to
-# hypervisors to which tunnels should be connected. It is best to use
-# an absolute path to this file.
-# remote-ip-file = /opt/stack/remote-ips.txt
 
 # Uncomment this line if enable-tunneling is True above.
 # Set local-ip to be the local IP address of this hypervisor.

quantum/plugins/openvswitch/agent/ovs_quantum_agent.py

@@ -28,7 +28,10 @@ import subprocess
 import sys
 import time
 
-from sqlalchemy.ext.sqlsoup import SqlSoup
+import sqlalchemy
+from sqlalchemy.ext import sqlsoup
+
+from quantum.plugins.openvswitch import ovs_models
 
 logging.basicConfig()
 LOG = logging.getLogger(__name__)
@@ -100,6 +103,10 @@ class OVSBridge:
         full_args = ["ovs-ofctl", cmd, self.br_name] + args
         return self.run_cmd(full_args)
 
+    def count_flows(self):
+        flow_list = self.run_ofctl("dump-flows", []).split("\n")[1:]
+        return len(flow_list) - 1
+
     def remove_all_flows(self):
         self.run_ofctl("del-flows", [])
 
@@ -214,6 +221,36 @@ class LocalVLANMapping:
         return "lv-id = %s ls-id = %s" % (self.vlan, self.lsw_id)
 
 
+class Port(object):
+    '''class stores port data in an ORM-free way,
+       so attributes are still available even if a
+       row has been deleted.
+    '''
+
+    def __init__(self, p):
+        self.uuid = p.uuid
+        self.network_id = p.network_id
+        self.interface_id = p.interface_id
+        self.state = p.state
+        self.op_status = p.op_status
+
+    def __eq__(self, other):
+        '''compare only fields that will cause us to re-wire
+        '''
+        try:
+            return (self and other
+                    and self.interface_id == other.interface_id
+                    and self.state == other.state)
+        except:
+            return False
+
+    def __ne__(self, other):
+        return not self.__eq__(other)
+
+    def __hash__(self):
+        return hash(self.uuid)
+
+
 class OVSQuantumAgent(object):
 
     def __init__(self, integ_br, root_helper,
@@ -239,6 +276,10 @@ class OVSQuantumAgent(object):
         self.int_br.add_flow(priority=1, actions="normal")
 
     def daemon_loop(self, db_connection_url):
+        '''Main processing loop for Non-Tunneling Agent.
+
+        :param options: database information - in the event need to reconnect
+        '''
         self.local_vlan_map = {}
         old_local_bindings = {}
         old_vif_ports = {}
@@ -247,7 +288,7 @@ class OVSQuantumAgent(object):
         while True:
             if not db_connected:
                 time.sleep(self.reconnect_interval)
-                db = SqlSoup(db_connection_url)
+                db = sqlsoup.SqlSoup(db_connection_url)
                 db_connected = True
                 LOG.info("Connecting to database \"%s\" on %s" %
                          (db.engine.url.database, db.engine.url.host))
@@ -358,25 +399,30 @@ class OVSQuantumTunnelAgent(object):
     # Upper bound on available vlans.
     MAX_VLAN_TAG = 4094
 
-    def __init__(self, integ_br, tun_br, remote_ip_file, local_ip,
+    def __init__(self, integ_br, tun_br, local_ip, root_helper,
-                 root_helper, polling_interval, reconnect_interval):
+                 polling_interval, reconnect_interval):
         '''Constructor.
 
         :param integ_br: name of the integration bridge.
         :param tun_br: name of the tunnel bridge.
-        :param remote_ip_file: name of file containing list of hypervisor IPs.
+        :param local_ip: local IP address of this hypervisor.
-        :param local_ip: local IP address of this hypervisor.'''
+        :param root_helper: utility to use when running shell cmds.
+        :param polling_interval: interval (secs) to poll DB.
+        :param reconnect_internal: retry interval (secs) on DB error.'''
         self.root_helper = root_helper
         self.available_local_vlans = set(
             xrange(OVSQuantumTunnelAgent.MIN_VLAN_TAG,
                    OVSQuantumTunnelAgent.MAX_VLAN_TAG))
         self.setup_integration_br(integ_br)
         self.local_vlan_map = {}
-        self.setup_tunnel_br(tun_br, remote_ip_file, local_ip)
+
-        self.db_connected = False
         self.polling_interval = polling_interval
         self.reconnect_interval = reconnect_interval
 
+        self.local_ip = local_ip
+        self.tunnel_count = 0
+        self.setup_tunnel_br(tun_br)
+
     def provision_local_vlan(self, net_uuid, lsw_id):
         '''Provisions a local VLAN.
 
@@ -391,8 +437,8 @@ class OVSQuantumTunnelAgent(object):
         # outbound
         self.tun_br.add_flow(priority=4, match="in_port=%s,dl_vlan=%s" %
                              (self.patch_int_ofport, lvid),
-                             actions="set_tunnel:%s,normal" % (lsw_id))
+                             actions="strip_vlan,set_tunnel:%s,normal" %
-
+                             (lsw_id))
         # inbound
         self.tun_br.add_flow(priority=3, match="tun_id=%s" % lsw_id,
                              actions="mod_vlan_vid:%s,output:%s" %
@@ -471,97 +517,72 @@ class OVSQuantumTunnelAgent(object):
         # switch all traffic using L2 learning
         self.int_br.add_flow(priority=1, actions="normal")
 
-    def setup_tunnel_br(self, tun_br, remote_ip_file, local_ip):
+    def setup_tunnel_br(self, tun_br):
         '''Setup the tunnel bridge.
 
-        Reads in list of IP addresses. Creates GRE tunnels to each of these
+        Creates tunnel bridge, and links it to the integration bridge
-        addresses and then clears out existing flows. local_ip is the address
+        using a patch port.
-        of the local node. A tunnel is not created to this IP address.
 
-        :param tun_br: the name of the tunnel bridge.
+        :param tun_br: the name of the tunnel bridge.'''
-        :param remote_ip_file: path to file that contains list of destination
-            IP addresses.
-        :param local_ip: the ip address of this node.'''
         self.tun_br = OVSBridge(tun_br, self.root_helper)
         self.tun_br.reset_bridge()
         self.patch_int_ofport = self.tun_br.add_patch_port("patch-int",
                                                            "patch-tun")
-        try:
-            with open(remote_ip_file, 'r') as f:
-                remote_ip_list = f.readlines()
-                clean_ips = (x.rstrip() for x in remote_ip_list)
-                tunnel_ips = (x for x in clean_ips if x != local_ip and x)
-                for i, remote_ip in enumerate(tunnel_ips):
-                    self.tun_br.add_tunnel_port("gre-" + str(i), remote_ip)
-        except Exception as e:
-            LOG.error("Error configuring tunnels: '%s' %s" %
-                      (remote_ip_file, str(e)))
-            raise
-
         self.tun_br.remove_all_flows()
-        # default drop
         self.tun_br.add_flow(priority=1, actions="drop")
 
-    def get_db_port_bindings(self, db):
+    def manage_tunnels(self, tunnel_ips, old_tunnel_ips, db):
-        '''Get database port bindings from central Quantum database.
+        if self.local_ip in tunnel_ips:
+            tunnel_ips.remove(self.local_ip)
+        else:
+            db.tunnel_ips.insert(ip_address=self.local_ip)
 
-        The central quantum database 'ovs_quantum' resides on the openstack
+        new_tunnel_ips = tunnel_ips - old_tunnel_ips
-        mysql server.
+        if new_tunnel_ips:
+            LOG.info("adding tunnels to: %s" % new_tunnel_ips)
+            for ip in new_tunnel_ips:
+                tun_name = "gre-" + str(self.tunnel_count)
+                self.tun_br.add_tunnel_port(tun_name, ip)
+                self.tunnel_count += 1
 
-        :returns: a dictionary containing port bindings.'''
+            # adding new ports can void flows, so reset flows
-        ports = []
+            self.tun_br.remove_all_flows()
+            self.tun_br.add_flow(priority=1, actions="drop")
+            for lv_ojb in self.local_vlan_map.values():
+                self.add_tun_br_flows_for_local_vlan(lv_obj)
+
+    def rollback_until_success(self, db):
+        while True:
+            time.sleep(self.reconnect_interval)
             try:
-            ports = db.ports.all()
+                db.rollback()
-        except Exceptioni as e:
+                break
-            LOG.info("Unable to get port bindings! Exception: %s" % e)
+            except:
-            self.db_connected = False
+                LOG.exception("Problem connecting to database")
-            return {}
-
-        return dict([(port.interface_id, port) for port in ports])
-
-    def get_db_vlan_bindings(self, db):
-        '''Get database vlan bindings from central Quantum database.
-
-        The central quantum database 'ovs_quantum' resides on the openstack
-        mysql server.
-
-        :returns: a dictionary containing vlan bindings.'''
-        lsw_id_binds = []
-        try:
-            lsw_id_binds.extend(db.vlan_bindings.all())
-        except Exception as e:
-            LOG.info("Unable to get vlan bindings! Exception: %s" % e)
-            self.db_connected = False
-            return {}
-
-        return dict([(bind.network_id, bind.vlan_id)
-            for bind in lsw_id_binds])
 
     def daemon_loop(self, db_connection_url):
-        '''Main processing loop (not currently used).
+        '''Main processing loop for Tunneling Agent.
 
         :param options: database information - in the event need to reconnect
         '''
         old_local_bindings = {}
         old_vif_ports = {}
-        self.db_connected = False
+        old_tunnel_ips = set()
 
-        while True:
+        db = sqlsoup.SqlSoup(db_connection_url)
-            if not self.db_connected:
-                time.sleep(self.reconnect_interval)
-                db = SqlSoup(db_connection_url)
-                self.db_connected = True
         LOG.info("Connecting to database \"%s\" on %s" %
                  (db.engine.url.database, db.engine.url.host))
 
-            # Get bindings from db.
+        while True:
-            all_bindings = self.get_db_port_bindings(db)
+            try:
-            if not self.db_connected:
+                all_bindings = dict((p.interface_id, Port(p))
-                continue
+                                    for p in db.ports.all())
-            all_bindings_vif_port_ids = set(all_bindings.keys())
+                all_bindings_vif_port_ids = set(all_bindings)
-            lsw_id_bindings = self.get_db_vlan_bindings(db)
+                lsw_id_bindings = dict((bind.network_id, bind.vlan_id)
-            if not self.db_connected:
+                                       for bind in db.vlan_bindings.all())
-                continue
+
+                tunnel_ips = set(x.ip_address for x in db.tunnel_ips.all())
+                self.manage_tunnels(tunnel_ips, old_tunnel_ips, db)
 
                 # Get bindings from OVS bridge.
                 vif_ports = self.int_br.get_vif_ports()
@@ -569,27 +590,31 @@ class OVSQuantumTunnelAgent(object):
                 new_vif_ports_ids = set(new_vif_ports.keys())
 
                 old_vif_ports_ids = set(old_vif_ports.keys())
-            dead_vif_ports_ids = new_vif_ports_ids - all_bindings_vif_port_ids
+                dead_vif_ports_ids = (new_vif_ports_ids -
+                                      all_bindings_vif_port_ids)
                 dead_vif_ports = [new_vif_ports[p] for p in dead_vif_ports_ids]
-            disappeared_vif_ports_ids = old_vif_ports_ids - new_vif_ports_ids
+                disappeared_vif_ports_ids = (old_vif_ports_ids -
-            new_local_bindings_ids = all_bindings_vif_port_ids.intersection(
                                              new_vif_ports_ids)
+                new_local_bindings_ids = (all_bindings_vif_port_ids.
+                                          intersection(new_vif_ports_ids))
                 new_local_bindings = dict([(p, all_bindings.get(p))
                                            for p in new_vif_ports_ids])
                 new_bindings = set(
                     (p, old_local_bindings.get(p),
                      new_local_bindings.get(p)) for p in new_vif_ports_ids)
-            changed_bindings = set([b for b in new_bindings if b[2] != b[1]])
+                changed_bindings = set([b for b in new_bindings
+                                        if b[2] != b[1]])
 
-            LOG.debug('all_bindings: %s' % all_bindings)
+                LOG.debug('all_bindings: %s', all_bindings)
-            LOG.debug('lsw_id_bindings: %s' % lsw_id_bindings)
+                LOG.debug('lsw_id_bindings: %s', lsw_id_bindings)
-            LOG.debug('old_vif_ports_ids: %s' % old_vif_ports_ids)
+                LOG.debug('new_vif_ports_ids: %s', new_vif_ports_ids)
-            LOG.debug('dead_vif_ports_ids: %s' % dead_vif_ports_ids)
+                LOG.debug('dead_vif_ports_ids: %s', dead_vif_ports_ids)
-            LOG.debug('old_vif_ports_ids: %s' % old_vif_ports_ids)
+                LOG.debug('old_vif_ports_ids: %s', old_vif_ports_ids)
-            LOG.debug('new_local_bindings_ids: %s' % new_local_bindings_ids)
+                LOG.debug('new_local_bindings_ids: %s',
-            LOG.debug('new_local_bindings: %s' % new_local_bindings)
+                          new_local_bindings_ids)
-            LOG.debug('new_bindings: %s' % new_bindings)
+                LOG.debug('new_local_bindings: %s', new_local_bindings)
-            LOG.debug('changed_bindings: %s' % changed_bindings)
+                LOG.debug('new_bindings: %s', new_bindings)
+                LOG.debug('changed_bindings: %s', changed_bindings)
 
                 # Take action.
                 for p in dead_vif_ports:
@@ -606,6 +631,7 @@ class OVSQuantumTunnelAgent(object):
                                  old_net_uuid + " for " + str(p)
                                  + " added to dead vlan")
                         self.port_unbound(p, old_net_uuid)
+                        all_bindings[p.vif_id].op_status = OP_STATUS_DOWN
                         if not new_port:
                             self.port_dead(p)
 
@@ -617,38 +643,33 @@ class OVSQuantumTunnelAgent(object):
                             continue
 
                         lsw_id = lsw_id_bindings[new_net_uuid]
-                    try:
                         self.port_bound(p, new_net_uuid, lsw_id)
+                        all_bindings[p.vif_id].op_status = OP_STATUS_UP
                         LOG.info("Port " + str(p) + " on net-id = "
                                  + new_net_uuid + " bound to " +
                                  str(self.local_vlan_map[new_net_uuid]))
-                    except Exception as e:
-                        LOG.info("Unable to bind Port " + str(p) +
-                                 " on netid = " + new_net_uuid + " to "
-                                 + str(self.local_vlan_map[new_net_uuid]))
 
                 for vif_id in disappeared_vif_ports_ids:
                     LOG.info("Port Disappeared: " + vif_id)
+                    if vif_id in all_bindings:
+                        all_bindings[vif_id].op_status = OP_STATUS_DOWN
                     old_port = old_local_bindings.get(vif_id)
                     if old_port:
-                    try:
                         self.port_unbound(old_vif_ports[vif_id],
                                           old_port.network_id)
-                    except Exception:
+                # commit any DB changes and expire
-                        LOG.info("Unable to unbind Port " + str(p) +
+                # data loaded from the database
-                                 " on net-id = " + old_port.network_uuid)
+                db.commit()
 
+                # sleep and re-initialize state for next pass
+                time.sleep(self.polling_interval)
+                old_tunnel_ips = tunnel_ips
                 old_vif_ports = new_vif_ports
                 old_local_bindings = new_local_bindings
-            try:
-                db.commit()
-            except Exception as e:
-                LOG.info("Unable to commit to database! Exception: %s" % e)
-                db.rollback()
-                old_local_bindings = {}
-                old_vif_ports = {}
 
-            time.sleep(self.polling_interval)
+            except:
+                LOG.exception("Main-loop Exception:")
+                self.rollback_until_success(db)
 
 
 def main():
@@ -721,12 +742,6 @@ def main():
                 raise Exception('Empty tunnel-bridge in configuration file.')
 
             # Mandatory parameter.
-            remote_ip_file = config.get("OVS", "remote-ip-file")
-            if not len(remote_ip_file):
-                raise Exception('Empty remote-ip-file in configuration file.')
-
-            # Mandatory parameter.
-            remote_ip_file = config.get("OVS", "remote-ip-file")
             local_ip = config.get("OVS", "local-ip")
             if not len(local_ip):
                 raise Exception('Empty local-ip in configuration file.')
@@ -736,8 +751,7 @@ def main():
                       (config_file, str(e)))
             sys.exit(1)
 
-        plugin = OVSQuantumTunnelAgent(integ_br, tun_br, remote_ip_file,
+        plugin = OVSQuantumTunnelAgent(integ_br, tun_br, local_ip, root_helper,
-                                       local_ip, root_helper,
                                        polling_interval, reconnect_interval)
     else:
         # Get parameters for OVSQuantumAgent.

quantum/plugins/openvswitch/ovs_models.py

@@ -40,3 +40,16 @@ class VlanBinding(BASE):
     def __repr__(self):
         return "<VlanBinding(%s,%s)>" % \
           (self.vlan_id, self.network_id)
+
+
+class TunnelIP(BASE):
+    """Represents a remote IP in tunnel mode"""
+    __tablename__ = 'tunnel_ips'
+
+    ip_address = Column(String(255), primary_key=True)
+
+    def __init__(self, ip_address):
+        self.ip_address = ip_address
+
+    def __repr__(self):
+        return "<TunnelIP(%s)>" % (self.ip_address)

quantum/plugins/openvswitch/tests/unit/test_tunnel.py

@@ -24,9 +24,6 @@ import mox
 from quantum.plugins.openvswitch.agent import ovs_quantum_agent
 
 
-LOCAL_DIR = os.path.dirname(__file__)
-REMOTE_IP_FILE = LOCAL_DIR + '/remote-ip-file.txt'
-
 # Useful global dummy variables.
 NET_UUID = '3faeebfe-5d37-11e1-a64b-000c29d5f0a7'
 LS_ID = '42'
@@ -53,7 +50,6 @@ class DummyVlanBinding:
 class TunnelTest(unittest.TestCase):
 
     def setUp(self):
-        print LOCAL_DIR
         self.mox = mox.Mox()
 
         self.INT_BRIDGE = 'integration_bridge'
@@ -86,7 +82,6 @@ class TunnelTest(unittest.TestCase):
 
         b = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         self.mox.VerifyAll()
@@ -106,7 +101,6 @@ class TunnelTest(unittest.TestCase):
 
         a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         a.available_local_vlans = set([LV_ID])
@@ -123,7 +117,6 @@ class TunnelTest(unittest.TestCase):
         self.mox.ReplayAll()
         a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         a.available_local_vlans = set()
@@ -140,7 +133,6 @@ class TunnelTest(unittest.TestCase):
         self.mox.ReplayAll()
         a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         a.local_vlan_map[NET_UUID] = LVM
@@ -151,7 +143,6 @@ class TunnelTest(unittest.TestCase):
         self.mox.ReplayAll()
         a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         a.available_local_vlans = set([LV_ID])
@@ -170,41 +161,9 @@ class TunnelTest(unittest.TestCase):
         self.mox.ReplayAll()
         a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
                                                     self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
                                                     '10.0.0.1',
                                                     'sudo', 2, 2)
         a.available_local_vlans = set([LV_ID])
         a.local_vlan_map[NET_UUID] = LVM
         a.port_dead(VIF_PORT)
         self.mox.VerifyAll()
-
-    def testDbBindings(self):
-        db = self.mox.CreateMockAnything()
-        db.ports = self.mox.CreateMockAnything()
-        interface_ids = ['interface-id-%d' % x for x in range(3)]
-        db.ports.all().AndReturn([DummyPort(x) for x in interface_ids])
-
-        db.vlan_bindings = self.mox.CreateMockAnything()
-        vlan_bindings = [
-            ['network-id-%d' % x, 'vlan-id-%d' % x] for x in range(3)]
-        db.vlan_bindings.all().AndReturn(
-            [DummyVlanBinding(*x) for x in vlan_bindings])
-
-        self.mox.ReplayAll()
-        a = ovs_quantum_agent.OVSQuantumTunnelAgent(self.INT_BRIDGE,
-                                                    self.TUN_BRIDGE,
-                                                    REMOTE_IP_FILE,
-                                                    '10.0.0.1',
-                                                    'sudo', 2, 2)
-
-        all_bindings = a.get_db_port_bindings(db)
-        lsw_id_bindings = a.get_db_vlan_bindings(db)
-
-        for interface_id, port in all_bindings.iteritems():
-            self.assertTrue(interface_id in interface_ids)
-
-        for network_id, vlan_id in lsw_id_bindings.iteritems():
-            self.assertTrue(network_id in [x[0] for x in vlan_bindings])
-            self.assertTrue(vlan_id in [x[1] for x in vlan_bindings])
-
-        self.mox.VerifyAll()