Merge pull request #61 from stackhpc/upstream/yoga-2023-08-14

Synchronise yoga with upstream

Author: markgoddard

neutron/cmd/ovn/neutron_ovn_db_sync_util.py

@@ -57,12 +57,12 @@ def post_fork_initialize(self, resource, event, trigger, **kwargs):
     def ovn_client(self):
         return self._ovn_client
 
-    def _set_hash_ring_nodes_offline(self):
-        """Don't set hash ring nodes as offline.
+    def _remove_node_from_hash_ring(self):
+        """Don't remove the node from the Hash Ring.
 
         If this method was not overridden, cleanup would be performed when
-        calling the db sync and running neutron server would mark all the
-        nodes from the ring as offline.
+        calling the db sync and running neutron server would remove the
+        nodes from the Hash Ring.
         """
 
     # Since we are not using the ovn mechanism driver while syncing,

neutron/common/utils.py

@@ -35,9 +35,12 @@
 from eventlet.green import subprocess
 import netaddr
 from neutron_lib.api.definitions import availability_zone as az_def
+from neutron_lib.api.definitions import portbindings
+from neutron_lib.api.definitions import portbindings_extended
 from neutron_lib import constants as n_const
 from neutron_lib import context as n_context
 from neutron_lib.db import api as db_api
+from neutron_lib.plugins import utils as plugin_utils
 from neutron_lib.services.trunk import constants as trunk_constants
 from neutron_lib.utils import helpers
 from oslo_config import cfg
@@ -1046,3 +1049,16 @@ def is_session_active(session):
     if not (session.dirty or session.deleted or session.new):
         return False
     return True
+
+
+# TODO(slaweq): this should be moved to neutron_lib.plugins.utils module
+def is_port_bound(port, log_message=True):
+    active_binding = plugin_utils.get_port_binding_by_status_and_host(
+        port.get('port_bindings', []), n_const.ACTIVE)
+    if not active_binding:
+        if log_message:
+            LOG.warning('Binding for port %s was not found.', port)
+        return False
+    return active_binding[portbindings_extended.VIF_TYPE] not in (
+        portbindings.VIF_TYPE_UNBOUND,
+        portbindings.VIF_TYPE_BINDING_FAILED)

neutron/db/l3_dvr_db.py

@@ -17,7 +17,6 @@
 from neutron_lib.api.definitions import external_net as extnet_apidef
 from neutron_lib.api.definitions import l3 as l3_apidef
 from neutron_lib.api.definitions import portbindings
-from neutron_lib.api.definitions import portbindings_extended
 from neutron_lib.api.definitions import router_admin_state_down_before_update
 from neutron_lib.api import validators
 from neutron_lib.callbacks import events
@@ -70,18 +69,6 @@ def is_admin_state_down_necessary():
     return _IS_ADMIN_STATE_DOWN_NECESSARY
 
 
-# TODO(slaweq): this should be moved to neutron_lib.plugins.utils module
-def is_port_bound(port):
-    active_binding = plugin_utils.get_port_binding_by_status_and_host(
-        port.get("port_bindings", []), const.ACTIVE)
-    if not active_binding:
-        LOG.warning("Binding for port %s was not found.", port)
-        return False
-    return active_binding[portbindings_extended.VIF_TYPE] not in [
-        portbindings.VIF_TYPE_UNBOUND,
-        portbindings.VIF_TYPE_BINDING_FAILED]
-
-
 @registry.has_registry_receivers
 class DVRResourceOperationHandler(object):
     """Contains callbacks for DVR operations.
@@ -1426,7 +1413,7 @@ def is_router_distributed(self, context, router_id):
 
     def get_ports_under_dvr_connected_subnet(self, context, subnet_id):
         query = dvr_mac_db.get_ports_query_by_subnet_and_ip(context, subnet_id)
-        ports = [p for p in query.all() if is_port_bound(p)]
+        ports = [p for p in query.all() if n_utils.is_port_bound(p)]
         # TODO(slaweq): if there would be way to pass to neutron-lib only
         # list of extensions which actually should be processed, than setting
         # process_extensions=True below could avoid that second loop and

neutron/db/ovn_hash_ring_db.py

@@ -50,6 +50,13 @@ def remove_nodes_from_host(context, group_name):
              CONF.host, group_name)
 
 
+def remove_node_by_uuid(context, node_uuid):
+    with db_api.CONTEXT_WRITER.using(context):
+        context.session.query(ovn_models.OVNHashRing).filter(
+            ovn_models.OVNHashRing.node_uuid == node_uuid).delete()
+    LOG.info('Node "%s" removed from the Hash Ring', node_uuid)
+
+
 def _touch(context, updated_at=None, **filter_args):
     if updated_at is None:
         updated_at = timeutils.utcnow()
@@ -96,7 +103,9 @@ def count_offline_nodes(context, interval, group_name):
     return query.count()
 
 
-def set_nodes_from_host_as_offline(context, group_name):
-    timestamp = datetime.datetime(day=26, month=10, year=1985, hour=9)
-    _touch(context, updated_at=timestamp, hostname=CONF.host,
-           group_name=group_name)
+@db_api.CONTEXT_READER
+def count_nodes_from_host(context, group_name):
+    query = context.session.query(ovn_models.OVNHashRing).filter(
+        ovn_models.OVNHashRing.group_name == group_name,
+        ovn_models.OVNHashRing.hostname == CONF.host)
+    return query.count()

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

@@ -270,17 +270,17 @@ def subscribe(self):
                                resources.SECURITY_GROUP_RULE,
                                events.BEFORE_DELETE)
 
-    def _set_hash_ring_nodes_offline(self, *args, **kwargs):
+    def _remove_node_from_hash_ring(self, *args, **kwargs):
+        # The node_uuid attribute will be empty for worker types
+        # that are not added to the Hash Ring and can be skipped
+        if self.node_uuid is None:
+            return
         admin_context = n_context.get_admin_context()
-        ovn_hash_ring_db.set_nodes_from_host_as_offline(
-            admin_context, self.hash_ring_group)
-        LOG.info('Hash Ring nodes from host "%s" marked as offline',
-                 cfg.CONF.host)
+        ovn_hash_ring_db.remove_node_by_uuid(
+            admin_context, self.node_uuid)
 
     def pre_fork_initialize(self, resource, event, trigger, payload=None):
         """Pre-initialize the ML2/OVN driver."""
-        atexit.register(self._set_hash_ring_nodes_offline)
-        signal.signal(signal.SIGTERM, self._set_hash_ring_nodes_offline)
         ovn_utils.create_neutron_pg_drop()
 
     @staticmethod
@@ -298,6 +298,10 @@ def _setup_hash_ring(self):
         thread for this host. Subsequently workers just need to register
         themselves to the hash ring.
         """
+        # Attempt to remove the node from the ring when the worker stops
+        atexit.register(self._remove_node_from_hash_ring)
+        signal.signal(signal.SIGTERM, self._remove_node_from_hash_ring)
+
         admin_context = n_context.get_admin_context()
         if not self._hash_ring_probe_event.is_set():
             # Clear existing entries

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

@@ -44,6 +44,7 @@
 from neutron.objects import router as router_obj
 from neutron.objects import ports as ports_obj
 from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovn_db_sync
+from neutron import service
 from neutron.services.logapi.drivers.ovn import driver as log_driver
 
 
@@ -1123,3 +1124,20 @@ def touch_hash_ring_nodes(self):
         # here because we want the maintenance tasks from each instance to
         # execute this task.
         hash_ring_db.touch_nodes_from_host(self.ctx, self._group)
+
+        # Check the number of the nodes in the ring and log a message in
+        # case they are out of sync. See LP #2024205 for more information
+        # on this issue.
+        api_workers = service._get_api_workers()
+        num_nodes = hash_ring_db.count_nodes_from_host(self.ctx, self._group)
+
+        if num_nodes > api_workers:
+            LOG.critical(
+                'The number of nodes in the Hash Ring (%d) is higher than '
+                'the number of API workers (%d) for host "%s". Something is '
+                'not right and OVSDB events could be missed because of this. '
+                'Please check the status of the Neutron processes, this can '
+                'happen when the API workers are killed and restarted. '
+                'Restarting the service should fix the issue, see LP '
+                '#2024205 for more information.',
+                num_nodes, api_workers, cfg.CONF.host)

neutron/services/trunk/drivers/ovn/trunk_driver.py

@@ -22,10 +22,12 @@
 from oslo_log import log
 
 from neutron.common.ovn import constants as ovn_const
+from neutron.common import utils as n_utils
 from neutron.db import db_base_plugin_common
 from neutron.db import ovn_revision_numbers_db as db_rev
 from neutron.objects import ports as port_obj
 from neutron.services.trunk.drivers import base as trunk_base
+from neutron.services.trunk import exceptions as trunk_exc
 
 
 SUPPORTED_INTERFACES = (
@@ -162,6 +164,10 @@ def _unset_binding_profile(self, context, subport, ovn_txn):
         LOG.debug("Done unsetting parent for subport %s", subport.port_id)
         return db_port
 
+    @staticmethod
+    def _is_port_bound(port):
+        return n_utils.is_port_bound(port, log_message=False)
+
     def trunk_updated(self, trunk):
         # Check if parent port is handled by OVN.
         if not self.plugin_driver.nb_ovn.lookup('Logical_Switch_Port',
@@ -208,6 +214,16 @@ def trunk_event(self, resource, event, trunk_plugin, payload):
             self.trunk_updated(payload.states[0])
         elif event == events.AFTER_DELETE:
             self.trunk_deleted(payload.states[0])
+        elif event == events.PRECOMMIT_CREATE:
+            trunk = payload.desired_state
+            parent_port = trunk.db_obj.port
+            if self._is_port_bound(parent_port):
+                raise trunk_exc.ParentPortInUse(port_id=parent_port.id)
+        elif event == events.PRECOMMIT_DELETE:
+            trunk = payload.states[0]
+            parent_port = payload.states[1]
+            if self._is_port_bound(parent_port):
+                raise trunk_exc.TrunkInUse(trunk_id=trunk.id)
 
     def subport_event(self, resource, event, trunk_plugin, payload):
         if event == events.AFTER_CREATE:
@@ -233,7 +249,8 @@ def register(self, resource, event, trigger, payload=None):
             resource, event, trigger, payload=payload)
         self._handler = OVNTrunkHandler(self.plugin_driver)
         for _event in (events.AFTER_CREATE, events.AFTER_UPDATE,
-                       events.AFTER_DELETE):
+                       events.AFTER_DELETE, events.PRECOMMIT_CREATE,
+                       events.PRECOMMIT_DELETE):
             registry.subscribe(self._handler.trunk_event,
                                resources.TRUNK,
                                _event)

neutron/services/trunk/plugin.py

@@ -294,6 +294,7 @@ def delete_trunk(self, context, trunk_id):
             trunk = self._get_trunk(context, trunk_id)
             rules.trunk_can_be_managed(context, trunk)
             trunk_port_validator = rules.TrunkPortValidator(trunk.port_id)
+            parent_port = trunk.db_obj.port
             if trunk_port_validator.can_be_trunked_or_untrunked(context):
                 # NOTE(status_police): when a trunk is deleted, the logical
                 # object disappears from the datastore, therefore there is no
@@ -307,7 +308,7 @@ def delete_trunk(self, context, trunk_id):
                                     'deleting trunk port %s: %s', trunk_id,
                                     str(e))
                 payload = events.DBEventPayload(context, resource_id=trunk_id,
-                                                states=(trunk,))
+                                                states=(trunk, parent_port))
                 registry.publish(resources.TRUNK, events.PRECOMMIT_DELETE,
                                  self, payload=payload)
             else:
@@ -317,7 +318,7 @@ def delete_trunk(self, context, trunk_id):
         registry.publish(resources.TRUNK, events.AFTER_DELETE, self,
                          payload=events.DBEventPayload(
                              context, resource_id=trunk_id,
-                             states=(trunk,)))
+                             states=(trunk, parent_port)))
 
     @db_base_plugin_common.convert_result_to_dict
     def add_subports(self, context, trunk_id, subports):

neutron/tests/functional/base.py

@@ -363,7 +363,7 @@ def trigger(self):
         # NOTE(ralonsoh): do not access to the DB at exit when the SQL
         # connection is already closed, to avoid useless exception messages.
         mock.patch.object(
-            self.mech_driver, '_set_hash_ring_nodes_offline').start()
+            self.mech_driver, '_remove_node_from_hash_ring').start()
         self.mech_driver.pre_fork_initialize(
             mock.ANY, mock.ANY, trigger_cls.trigger)
 

neutron/tests/functional/services/trunk/drivers/ovn/test_trunk_driver.py

@@ -15,6 +15,7 @@
 import contextlib
 
 from neutron_lib.api.definitions import portbindings
+from neutron_lib.callbacks import exceptions as n_exc
 from neutron_lib import constants as n_consts
 from neutron_lib.db import api as db_api
 from neutron_lib.plugins import utils
@@ -113,6 +114,25 @@ def test_trunk_create_with_subports(self):
             with self.trunk([subport]) as trunk:
                 self._verify_trunk_info(trunk, has_items=True)
 
+    def test_trunk_create_parent_port_bound(self):
+        with self.network() as network:
+            with self.subnet(network=network) as subnet:
+                with self.port(subnet=subnet) as parent_port:
+                    pb = port_obj.PortBinding.get_objects(
+                        self.context, port_id=parent_port['port']['id'])
+                    port_obj.PortBinding.update_object(
+                        self.context, {'vif_type': portbindings.VIF_TYPE_OVS},
+                        port_id=pb[0].port_id, host=pb[0].host)
+                    tenant_id = uuidutils.generate_uuid()
+                    trunk = {'trunk': {
+                        'port_id': parent_port['port']['id'],
+                        'tenant_id': tenant_id, 'project_id': tenant_id,
+                        'admin_state_up': True,
+                        'name': 'trunk', 'sub_ports': []}}
+                    self.assertRaises(n_exc.CallbackFailure,
+                                      self.trunk_plugin.create_trunk,
+                                      self.context, trunk)
+
     def test_subport_add(self):
         with self.subport() as subport:
             with self.trunk() as trunk:
@@ -147,3 +167,14 @@ def test_trunk_delete(self):
         with self.trunk() as trunk:
             self.trunk_plugin.delete_trunk(self.context, trunk['id'])
             self._verify_trunk_info({}, has_items=False)
+
+    def test_trunk_delete_parent_port_bound(self):
+        with self.trunk() as trunk:
+            bp = port_obj.PortBinding.get_objects(
+                self.context, port_id=trunk['port_id'])
+            port_obj.PortBinding.update_object(
+                self.context, {'vif_type': portbindings.VIF_TYPE_OVS},
+                port_id=bp[0].port_id, host=bp[0].host)
+            self.assertRaises(n_exc.CallbackFailure,
+                              self.trunk_plugin.delete_trunk,
+                              self.context, trunk['id'])