ovn: Use ovsdb-client to create neutron_pg_drop

Previously we used short living OVN database connection to create
neutron_pg_drop Port Group before workers were spawned. The
pre_fork_initialize actually happens after the api workers are spawned
anyways and it blocks spawning of other workers, such as maintenance,
rpc or periodic. If the OVN database was large it may take several
minutes to connect to the database at scale and this blocks spawning of
other workers. That means connecting to OVN in pre_fork is not a good
idea.

This patch replaces the mechanism by using ovsdb-client to send a
transaction without connecting to the database and downloading the whole
content. The command does following, everything is on the server side:

 1) With timeout 0 it waits for neutron_pg_drop Port Group. If the PG is
    present, the transaction finishes and nothing happens.

 2) If the PG is not present, it times out immediately and commits new
    entries that effectivelly creates neutron_pg_drop Port Group with
    implicit ACLs to block ingress and egress traffic.

Conflicts:
	neutron/plugins/ml2/drivers/ovn/mech_driver/mech_driver.py
	neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

Closes-Bug: #1991579
Co-Authored-By: Terry Wilson <[email protected]>

Change-Id: I27af495f96a3ea88dd31345dbfb55f1be8faabd6
(cherry picked from commit 50eee19)
(cherry picked from commit 5deea00)

Author: cubeek

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()