Merge "ovn: Use ovsdb-client to create neutron_pg_drop" into stable/yoga

Author: Zuul

neutron/common/ovn/utils.py

@@ -27,6 +27,7 @@
 from neutron_lib import exceptions as n_exc
 from neutron_lib.plugins import directory
 from neutron_lib.utils import net as n_utils
+from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_log import log
 from oslo_serialization import jsonutils
@@ -56,6 +57,70 @@
     'PortExtraDHCPValidation', ['failed', 'invalid_ipv4', 'invalid_ipv6'])
 
 
+class OvsdbClientCommand(object):
+    _CONNECTION = 0
+    _PRIVATE_KEY = 1
+    _CERTIFICATE = 2
+    _CA_AUTHORITY = 3
+
+    OVN_Northbound = "OVN_Northbound"
+    OVN_Southbound = "OVN_Southbound"
+
+    _db_settings = {
+        OVN_Northbound: {
+            _CONNECTION: ovn_conf.get_ovn_nb_connection,
+            _PRIVATE_KEY: ovn_conf.get_ovn_nb_private_key,
+            _CERTIFICATE: ovn_conf.get_ovn_nb_certificate,
+            _CA_AUTHORITY: ovn_conf.get_ovn_nb_ca_cert,
+        },
+        OVN_Southbound: {
+            _CONNECTION: ovn_conf.get_ovn_sb_connection,
+            _PRIVATE_KEY: ovn_conf.get_ovn_sb_private_key,
+            _CERTIFICATE: ovn_conf.get_ovn_sb_certificate,
+            _CA_AUTHORITY: ovn_conf.get_ovn_sb_ca_cert,
+        },
+    }
+
+    @classmethod
+    def run(cls, command):
+        """Run custom ovsdb protocol command.
+
+        :param command: JSON object of ovsdb protocol command
+        """
+        try:
+            db = command[0]
+        except IndexError:
+            raise KeyError(
+                _("%s or %s schema must be specified in the command %s" % (
+                    cls.OVN_Northbound, cls.OVN_Southbound, command)))
+
+        if db not in (cls.OVN_Northbound, cls.OVN_Southbound):
+            raise KeyError(
+                _("%s or %s schema must be specified in the command %s" % (
+                    cls.OVN_Northbound, cls.OVN_Southbound, command)))
+
+        cmd = ['ovsdb-client',
+               cls.COMMAND,
+               cls._db_settings[db][cls._CONNECTION](),
+               '--timeout',
+               str(ovn_conf.get_ovn_ovsdb_timeout())]
+
+        if cls._db_settings[db][cls._PRIVATE_KEY]():
+            cmd += ['-p', cls._db_settings[db][cls._PRIVATE_KEY](),
+                    '-c', cls._db_settings[db][cls._CERTIFICATE](),
+                    '-C', cls._db_settings[db][cls._CA_AUTHORITY]()]
+
+        cmd.append(jsonutils.dumps(command))
+
+        return processutils.execute(
+            *cmd,
+            log_errors=processutils.LOG_FINAL_ERROR)
+
+
+class OvsdbClientTransactCommand(OvsdbClientCommand):
+    COMMAND = 'transact'
+
+
 def ovn_name(id):
     # The name of the OVN entry will be neutron-<UUID>
     # This is due to the fact that the OVN application checks if the name
@@ -697,3 +762,56 @@ def wrapper(*args, **kwargs):
                 reraise=True)(func)(*args, **kwargs)
         return wrapper
     return inner
+
+
+def create_neutron_pg_drop():
+    """Create neutron_pg_drop Port Group.
+
+    It uses ovsdb-client to send to server transact command using ovsdb
+    protocol that checks if the neutron_pg_drop row exists. If it exists
+    it times out immediatelly. If it doesn't exist then it creates the
+    Port_Group and default ACLs to drop all ingress and egress traffic.
+    """
+    command = [
+        "OVN_Northbound", {
+            "op": "wait",
+            "timeout": 0,
+            "table": "Port_Group",
+            "where": [
+                ["name", "==", constants.OVN_DROP_PORT_GROUP_NAME]
+            ],
+            "until": "==",
+            "rows": []
+        }, {
+            "op": "insert",
+            "table": "ACL",
+            "row": {
+                "action": "drop",
+                "direction": "to-lport",
+                "match": "outport == @neutron_pg_drop && ip",
+                "priority": 1001
+            },
+            "uuid-name": "droptoport"
+        }, {
+            "op": "insert",
+            "table": "ACL",
+            "row": {
+                "action": "drop",
+                "direction": "from-lport",
+                "match": "inport == @neutron_pg_drop && ip",
+                "priority": 1001
+            },
+            "uuid-name": "dropfromport"
+        }, {
+            "op": "insert",
+            "table": "Port_Group",
+            "row": {
+                "name": constants.OVN_DROP_PORT_GROUP_NAME,
+                "acls": ["set", [
+                    ["named-uuid", "droptoport"],
+                    ["named-uuid", "dropfromport"]
+                ]]
+            }
+        }]
+
+    OvsdbClientTransactCommand.run(command)

neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py

@@ -36,7 +36,6 @@
 from neutron_lib.plugins.ml2 import api
 from neutron_lib.utils import helpers
 from oslo_concurrency import lockutils
-from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_db import exception as os_db_exc
 from oslo_log import log
@@ -59,7 +58,6 @@
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import maintenance
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovn_client
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovn_db_sync
-from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovsdb_monitor
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import worker
 from neutron import service
 from neutron.services.logapi.drivers.ovn import driver as log_driver
@@ -279,47 +277,7 @@ def pre_fork_initialize(self, resource, event, trigger, payload=None):
         """Pre-initialize the ML2/OVN driver."""
         atexit.register(self._clean_hash_ring)
         signal.signal(signal.SIGTERM, self._clean_hash_ring)
-        self._create_neutron_pg_drop()
-
-    def _create_neutron_pg_drop(self):
-        """Create neutron_pg_drop Port Group.
-
-        The method creates a short living connection to the Northbound
-        database. Because of multiple controllers can attempt to create the
-        Port Group at the same time the transaction can fail and raise
-        RuntimeError. In such case, we make sure the Port Group was created,
-        otherwise the error is something else and it's raised to the caller.
-        """
-        idl = ovsdb_monitor.OvnInitPGNbIdl.from_server(
-            ovn_conf.get_ovn_nb_connection(), self.nb_schema_helper, self)
-        # Only one server should try to create the port group
-        idl.set_lock('pg_drop_creation')
-        with ovsdb_monitor.short_living_ovsdb_api(
-                impl_idl_ovn.OvsdbNbOvnIdl, idl) as pre_ovn_nb_api:
-            try:
-                create_default_drop_port_group(pre_ovn_nb_api)
-            except KeyError:
-                # Due to a bug in python-ovs, we can send transactions before
-                # the initial OVSDB is populated in memory. This can break
-                # the AddCommand post_commit method which tries to return a
-                # row looked up by the newly commited row's uuid. Since we
-                # don't care about the return value from the PgAddCommand, we
-                # can just catch the KeyError and continue. This can be
-                # removed when the python-ovs bug is resolved.
-                pass
-            except RuntimeError as re:
-                # If we don't get the lock, and the port group didn't exist
-                # when we tried to create it, it might still have been
-                # created by another server and we just haven't gotten the
-                # update yet.
-                LOG.info("Waiting for Port Group %(pg)s to be created",
-                         {'pg': ovn_const.OVN_DROP_PORT_GROUP_NAME})
-                if not idl.neutron_pg_drop_event.wait():
-                    LOG.error("Port Group %(pg)s was not created in time",
-                              {'pg': ovn_const.OVN_DROP_PORT_GROUP_NAME})
-                    raise re
-                LOG.info("Porg Group %(pg)s was created by another server",
-                         {'pg': ovn_const.OVN_DROP_PORT_GROUP_NAME})
+        ovn_utils.create_neutron_pg_drop()
 
     @staticmethod
     def should_post_fork_initialize(worker_class):
@@ -1191,19 +1149,17 @@ def set_port_status_down(self, port_id):
 
     def delete_mac_binding_entries(self, external_ip):
         """Delete all MAC_Binding entries associated to this IP address"""
-        cmd = ['ovsdb-client', 'transact', ovn_conf.get_ovn_sb_connection(),
-               '--timeout', str(ovn_conf.get_ovn_ovsdb_timeout())]
-
-        if ovn_conf.get_ovn_sb_private_key():
-            cmd += ['-p', ovn_conf.get_ovn_sb_private_key(), '-c',
-                    ovn_conf.get_ovn_sb_certificate(), '-C',
-                    ovn_conf.get_ovn_sb_ca_cert()]
-
-        cmd += ['["OVN_Southbound", {"op": "delete", "table": "MAC_Binding", '
-                '"where": [["ip", "==", "%s"]]}]' % external_ip]
+        cmd = [
+            "OVN_Southbound", {
+                "op": "delete",
+                "table": "MAC_Binding",
+                "where": [
+                    ["ip", "==", external_ip]
+                ]
+            }
+        ]
 
-        return processutils.execute(*cmd,
-                                    log_errors=processutils.LOG_FINAL_ERROR)
+        return ovn_utils.OvsdbClientTransactCommand.run(cmd)
 
     def update_segment_host_mapping(self, host, phy_nets):
         """Update SegmentHostMapping in DB"""
@@ -1375,28 +1331,6 @@ def delete_agent(self, context, id, _driver=None):
         if_exists=True).execute(check_error=True)
 
 
-def create_default_drop_port_group(nb_idl):
-    pg_name = ovn_const.OVN_DROP_PORT_GROUP_NAME
-    if nb_idl.get_port_group(pg_name):
-        LOG.debug("Port Group %s already exists", pg_name)
-        return
-    with nb_idl.transaction(check_error=True) as txn:
-        # If drop Port Group doesn't exist yet, create it.
-        txn.add(nb_idl.pg_add(pg_name, acls=[], may_exist=True))
-        # Add ACLs to this Port Group so that all traffic is dropped.
-        acls = ovn_acl.add_acls_for_drop_port_group(pg_name)
-        for acl in acls:
-            txn.add(nb_idl.pg_acl_add(may_exist=True, **acl))
-
-        ports_with_pg = set()
-        for pg in nb_idl.get_sg_port_groups().values():
-            ports_with_pg.update(pg['ports'])
-
-        if ports_with_pg:
-            # Add the ports to the default Port Group
-            txn.add(nb_idl.pg_add_ports(pg_name, list(ports_with_pg)))
-
-
 def get_availability_zones(cls, context, _driver, filters=None, fields=None,
                            sorts=None, limit=None, marker=None,
                            page_reverse=False):

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovn_client.py

@@ -33,7 +33,6 @@
 from neutron_lib.plugins import utils as p_utils
 from neutron_lib.utils import helpers
 from neutron_lib.utils import net as n_net
-from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_log import log
 from oslo_utils import excutils
@@ -1676,16 +1675,16 @@ def delete_mac_binding_entries_by_mac(self, mac):
         is refer to patch:
         https://review.opendev.org/c/openstack/neutron/+/812805
         """
-        cmd = ['ovsdb-client', 'transact', ovn_conf.get_ovn_sb_connection(),
-               '--timeout', str(ovn_conf.get_ovn_ovsdb_timeout())]
-        if ovn_conf.get_ovn_sb_private_key():
-            cmd += ['-p', ovn_conf.get_ovn_sb_private_key(), '-c',
-                    ovn_conf.get_ovn_sb_certificate(), '-C',
-                    ovn_conf.get_ovn_sb_ca_cert()]
-        cmd += ['["OVN_Southbound", {"op": "delete", "table": "MAC_Binding", '
-                '"where": [["mac", "==", "%s"]]}]' % mac]
-        return processutils.execute(*cmd,
-                                    log_errors=processutils.LOG_FINAL_ERROR)
+        cmd = [
+            "OVN_Southbound", {
+                "op": "delete",
+                "table": "MAC_Binding",
+                "where": [
+                    ["mac", "==", mac]
+                ]
+            }
+        ]
+        return utils.OvsdbClientTransactCommand.run(cmd)
 
     def _delete_lrouter_port(self, context, port_id, router_id=None, txn=None):
         """Delete a logical router port."""

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovsdb_monitor.py

@@ -13,7 +13,6 @@
 #    under the License.
 
 import abc
-import contextlib
 import datetime
 
 from neutron_lib import context as neutron_context
@@ -581,17 +580,6 @@ def run(self, event, row, old):
         self.driver.delete_mac_binding_entries(row.external_ip)
 
 
-class NeutronPgDropPortGroupCreated(row_event.WaitEvent):
-    """WaitEvent for neutron_pg_drop Create event."""
-    def __init__(self, timeout=None):
-        table = 'Port_Group'
-        events = (self.ROW_CREATE,)
-        conditions = (('name', '=', ovn_const.OVN_DROP_PORT_GROUP_NAME),)
-        super(NeutronPgDropPortGroupCreated, self).__init__(
-            events, table, conditions, timeout=timeout)
-        self.event_name = 'PortGroupCreated'
-
-
 class OvnDbNotifyHandler(row_event.RowEventHandler):
     def __init__(self, driver):
         self.driver = driver
@@ -837,56 +825,6 @@ def post_connect(self):
              PortBindingChassisUpdateEvent(self.driver)])
 
 
-class OvnInitPGNbIdl(OvnIdl):
-    """Very limited OVN NB IDL.
-
-    This IDL is intended to be used only in initialization phase with short
-    living DB connections.
-    """
-
-    tables = ['Port_Group', 'Logical_Switch_Port', 'ACL']
-
-    def __init__(self, driver, remote, schema):
-        super(OvnInitPGNbIdl, self).__init__(driver, remote, schema)
-        self.set_table_condition(
-            'Port_Group', [['name', '==', ovn_const.OVN_DROP_PORT_GROUP_NAME]])
-        self.neutron_pg_drop_event = NeutronPgDropPortGroupCreated(
-                timeout=ovn_conf.get_ovn_ovsdb_timeout())
-        self.notify_handler.watch_event(self.neutron_pg_drop_event)
-
-    def notify(self, event, row, updates=None):
-        # Go ahead and process events even if the lock is contended so we can
-        # know that some other server has created the drop group
-        self.notify_handler.notify(event, row, updates)
-
-    @classmethod
-    def from_server(cls, connection_string, helper, driver, pg_only=False):
-        if pg_only:
-            helper.register_table('Port_Group')
-        else:
-            for table in cls.tables:
-                helper.register_table(table)
-
-        return cls(driver, connection_string, helper)
-
-
-@contextlib.contextmanager
-def short_living_ovsdb_api(api_class, idl):
-    """Context manager for short living connections to the database.
-
-    :param api_class: Class implementing the database calls
-                      (e.g. from the impl_idl module)
-    :param idl: An instance of IDL class (e.g. instance of OvnNbIdl)
-    """
-    conn = connection.Connection(
-        idl, timeout=ovn_conf.get_ovn_ovsdb_timeout())
-    api = api_class(conn)
-    try:
-        yield api
-    finally:
-        api.ovsdb_connection.stop()
-
-
 def _check_and_set_ssl_files(schema_name):
     if schema_name == 'OVN_Southbound':
         priv_key_file = ovn_conf.get_ovn_sb_private_key()