Merge pull request #108 from stackhpc/upstream/zed-2024-01-08

Synchronise zed with upstream

Author: mnasiadka

neutron/agent/metadata/driver.py

@@ -294,7 +294,13 @@ def spawn_monitored_metadata_proxy(cls, monitor, ns_name, port, conf,
         pm = cls._get_metadata_proxy_process_manager(uuid, conf,
                                                      ns_name=ns_name,
                                                      callback=callback)
-        pm.enable()
+        try:
+            pm.enable()
+        except exceptions.ProcessExecutionError as exec_err:
+            LOG.error("Encountered process execution error %(err)s while "
+                      "starting process in namespace %(ns)s",
+                      {"err": exec_err, "ns": ns_name})
+            return
         monitor.register(uuid, METADATA_SERVICE_NAME, pm)
         cls.monitors[router_id] = pm
 

neutron/agent/ovn/metadata/driver.py

@@ -205,7 +205,13 @@ def spawn_monitored_metadata_proxy(cls, monitor, ns_name, port, conf,
         pm = cls._get_metadata_proxy_process_manager(uuid, conf,
                                                      ns_name=ns_name,
                                                      callback=callback)
-        pm.enable()
+        try:
+            pm.enable()
+        except exceptions.ProcessExecutionError as exec_err:
+            LOG.error("Encountered process execution error %(err)s while "
+                      "starting process in namespace %(ns)s",
+                      {"err": exec_err, "ns": ns_name})
+            return
         monitor.register(uuid, METADATA_SERVICE_NAME, pm)
         cls.monitors[router_id] = pm
 

neutron/api/rpc/handlers/securitygroups_rpc.py

@@ -431,6 +431,10 @@ def _select_sg_ids_for_ports(self, context, ports):
                       for sg_id in p['security_group_ids']))
         return [(sg_id, ) for sg_id in sg_ids]
 
-    def _is_security_group_stateful(self, context, sg_id):
-        sg = self.rcache.get_resource_by_id(resources.SECURITYGROUP, sg_id)
-        return sg.stateful
+    def _get_sgs_stateful_flag(self, context, sg_ids):
+        sgs_stateful = {}
+        for sg_id in sg_ids:
+            sg = self.rcache.get_resource_by_id(resources.SECURITYGROUP, sg_id)
+            sgs_stateful[sg_id] = sg.stateful
+
+        return sgs_stateful

neutron/db/securitygroups_rpc_base.py

@@ -211,12 +211,10 @@ def security_group_info_for_ports(self, context, ports):
                     # this set will be serialized into a list by rpc code
                     remote_address_group_info[remote_ag_id][ethertype] = set()
             direction = rule_in_db['direction']
-            stateful = self._is_security_group_stateful(context,
-                                                        security_group_id)
             rule_dict = {
                 'direction': direction,
                 'ethertype': ethertype,
-                'stateful': stateful}
+            }
 
             for key in ('protocol', 'port_range_min', 'port_range_max',
                         'remote_ip_prefix', 'remote_group_id',
@@ -234,6 +232,13 @@ def security_group_info_for_ports(self, context, ports):
             if rule_dict not in sg_info['security_groups'][security_group_id]:
                 sg_info['security_groups'][security_group_id].append(
                     rule_dict)
+
+        # Populate the security group "stateful" flag in the SGs list of rules.
+        for sg_id, stateful in self._get_sgs_stateful_flag(
+                context, sg_info['security_groups'].keys()).items():
+            for rule in sg_info['security_groups'][sg_id]:
+                rule['stateful'] = stateful
+
         # Update the security groups info if they don't have any rules
         sg_ids = self._select_sg_ids_for_ports(context, ports)
         for (sg_id, ) in sg_ids:
@@ -427,13 +432,13 @@ def _select_sg_ids_for_ports(self, context, ports):
         """
         raise NotImplementedError()
 
-    def _is_security_group_stateful(self, context, sg_id):
-        """Return whether the security group is stateful or not.
+    def _get_sgs_stateful_flag(self, context, sg_id):
+        """Return the security groups stateful flag.
 
-        Return True if the security group associated with the given ID
-        is stateful, else False.
+        Returns a dictionary with the SG ID as key and the stateful flag:
+        {sg_1: True, sg_2: False, ...}
         """
-        return True
+        raise NotImplementedError()
 
 
 class SecurityGroupServerRpcMixin(SecurityGroupInfoAPIMixin,
@@ -530,5 +535,5 @@ def _select_ips_for_remote_address_group(self, context,
         return ips_by_group
 
     @db_api.retry_if_session_inactive()
-    def _is_security_group_stateful(self, context, sg_id):
-        return sg_obj.SecurityGroup.get_sg_by_id(context, sg_id).stateful
+    def _get_sgs_stateful_flag(self, context, sg_ids):
+        return sg_obj.SecurityGroup.get_sgs_stateful_flag(context, sg_ids)

neutron/objects/securitygroup.py

@@ -133,6 +133,13 @@ def get_bound_project_ids(cls, context, obj_id):
                                            security_group_ids=[obj_id])
         return {port.project_id for port in port_objs}
 
+    @classmethod
+    @db_api.CONTEXT_READER
+    def get_sgs_stateful_flag(cls, context, sg_ids):
+        query = context.session.query(cls.db_model.id, cls.db_model.stateful)
+        query = query.filter(cls.db_model.id.in_(sg_ids))
+        return dict(query.all())
+
 
 @base.NeutronObjectRegistry.register
 class DefaultSecurityGroup(base.NeutronDbObject):

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

@@ -1351,7 +1351,8 @@ def sync_hostname_and_physical_networks(self, ctx):
         LOG.debug('OVN-SB Sync hostname and physical networks started')
         host_phynets_map = self.ovn_api.get_chassis_hostname_and_physnets()
         current_hosts = set(host_phynets_map)
-        previous_hosts = segments_db.get_hosts_mapped_with_segments(ctx)
+        previous_hosts = segments_db.get_hosts_mapped_with_segments(
+            ctx, include_agent_types={ovn_const.OVN_CONTROLLER_AGENT})
 
         stale_hosts = previous_hosts - current_hosts
         for host in stale_hosts:

neutron/tests/functional/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_db_sync.py

@@ -1818,6 +1818,22 @@ def setUp(self):
     def _sync_resources(self):
         self.sb_synchronizer.sync_hostname_and_physical_networks(self.ctx)
 
+    def create_agent(self, host, bridge_mappings=None, agent_type=None):
+        if agent_type is None:
+            agent_type = ovn_const.OVN_CONTROLLER_AGENT
+        if bridge_mappings is None:
+            bridge_mappings = {}
+        agent = {
+            'host': host,
+            'agent_type': agent_type,
+            'binary': '/bin/test',
+            'topic': 'test_topic',
+            'configurations': {'bridge_mappings': bridge_mappings}
+        }
+        _, status = self.plugin.create_or_update_agent(self.context, agent)
+
+        return status['id']
+
     def create_segment(self, network_id, physical_network, segmentation_id):
         segment_data = {'network_id': network_id,
                         'physical_network': physical_network,
@@ -1858,6 +1874,7 @@ def test_ovn_sb_sync_delete_stale_host(self):
         segment = self.create_segment(network_id, 'physnet1', 50)
         segments_db.update_segment_host_mapping(
             self.ctx, 'host1', {segment['id']})
+        _ = self.create_agent('host1', bridge_mappings={'physnet1': 'eth0'})
         segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
         self.assertEqual({'host1'}, segment_hosts)
         # Since there is no chassis in the sb DB, host1 is the stale host
@@ -1866,6 +1883,36 @@ def test_ovn_sb_sync_delete_stale_host(self):
         segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
         self.assertFalse(segment_hosts)
 
+    def test_ovn_sb_sync_host_with_no_agent_not_deleted(self):
+        with self.network() as network:
+            network_id = network['network']['id']
+        segment = self.create_segment(network_id, 'physnet1', 50)
+        segments_db.update_segment_host_mapping(
+            self.ctx, 'host1', {segment['id']})
+        _ = self.create_agent('host1', bridge_mappings={'physnet1': 'eth0'},
+                              agent_type="Not OVN Agent")
+        segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
+        self.assertEqual({'host1'}, segment_hosts)
+        # There is no chassis in the sb DB, host1 does not have an agent
+        # so it is not deleted.
+        self._sync_resources()
+        segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
+        self.assertEqual({'host1'}, segment_hosts)
+
+    def test_ovn_sb_sync_host_with_other_agent_type_not_deleted(self):
+        with self.network() as network:
+            network_id = network['network']['id']
+        segment = self.create_segment(network_id, 'physnet1', 50)
+        segments_db.update_segment_host_mapping(
+            self.ctx, 'host1', {segment['id']})
+        segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
+        self.assertEqual({'host1'}, segment_hosts)
+        # There is no chassis in the sb DB, host1 does not have an agent
+        # so it is not deleted.
+        self._sync_resources()
+        segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
+        self.assertEqual({'host1'}, segment_hosts)
+
     def test_ovn_sb_sync(self):
         with self.network() as network:
             network_id = network['network']['id']
@@ -1878,6 +1925,9 @@ def test_ovn_sb_sync(self):
         segments_db.update_segment_host_mapping(
             self.ctx, 'host3', {seg1['id']})
         segment_hosts = segments_db.get_hosts_mapped_with_segments(self.ctx)
+        _ = self.create_agent('host1')
+        _ = self.create_agent('host2', bridge_mappings={'physnet2': 'eth0'})
+        _ = self.create_agent('host3', bridge_mappings={'physnet3': 'eth0'})
         self.assertEqual({'host1', 'host2', 'host3'}, segment_hosts)
         self.add_fake_chassis('host2', ['physnet2'])
         self.add_fake_chassis('host3', ['physnet3'])

neutron/tests/unit/agent/dhcp/test_agent.py

@@ -807,23 +807,37 @@ def _process_manager_constructor_call(self, ns=FAKE_NETWORK_DHCP_NS):
     def _enable_dhcp_helper(self, network, enable_isolated_metadata=False,
                             is_isolated_network=False, is_ovn_network=False):
         self.dhcp._process_monitor = mock.Mock()
+        # The disable() call
+        gmppm_expected_calls = [mock.call(FAKE_NETWORK_UUID, cfg.CONF,
+                                ns_name=FAKE_NETWORK_DHCP_NS)]
         if enable_isolated_metadata:
             cfg.CONF.set_override('enable_isolated_metadata', True)
+            if is_isolated_network:
+                # The enable() call
+                gmppm_expected_calls.append(
+                    mock.call(FAKE_NETWORK_UUID, cfg.CONF,
+                              ns_name=FAKE_NETWORK_DHCP_NS,
+                              callback=mock.ANY))
         self.plugin.get_network_info.return_value = network
-        self.dhcp.enable_dhcp_helper(network.id)
+        process_instance = mock.Mock(active=False)
+        with mock.patch.object(metadata_driver.MetadataDriver,
+                               '_get_metadata_proxy_process_manager',
+                               return_value=process_instance) as gmppm:
+            self.dhcp.enable_dhcp_helper(network.id)
+            gmppm.assert_has_calls(gmppm_expected_calls)
         self.plugin.assert_has_calls([
             mock.call.get_network_info(network.id)])
         self.call_driver.assert_called_once_with('enable', network)
         self.cache.assert_has_calls([mock.call.put(network)])
         if (is_isolated_network and enable_isolated_metadata and not
                 is_ovn_network):
-            self.external_process.assert_has_calls([
-                self._process_manager_constructor_call(),
-                mock.call().enable()], any_order=True)
+            process_instance.assert_has_calls([
+                mock.call.disable(sig=str(int(signal.SIGTERM))),
+                mock.call.get_pid_file_name(),
+                mock.call.enable()])
         else:
-            self.external_process.assert_has_calls([
-                self._process_manager_constructor_call(),
-                mock.call().disable(sig=str(int(signal.SIGTERM)))])
+            process_instance.assert_has_calls([
+                mock.call.disable(sig=str(int(signal.SIGTERM)))])
 
     def test_enable_dhcp_helper_enable_metadata_isolated_network(self):
         self._enable_dhcp_helper(isolated_network,
@@ -997,11 +1011,16 @@ def test_disable_dhcp_helper_driver_failure(self):
 
     def test_enable_isolated_metadata_proxy(self):
         self.dhcp._process_monitor = mock.Mock()
-        self.dhcp.enable_isolated_metadata_proxy(fake_network)
-        self.external_process.assert_has_calls([
-            self._process_manager_constructor_call(),
-            mock.call().enable()
-        ], any_order=True)
+        process_instance = mock.Mock(active=False)
+        with mock.patch.object(metadata_driver.MetadataDriver,
+                               '_get_metadata_proxy_process_manager',
+                               return_value=process_instance) as gmppm:
+            self.dhcp.enable_isolated_metadata_proxy(fake_network)
+            gmppm.assert_called_with(FAKE_NETWORK_UUID,
+                                     cfg.CONF,
+                                     ns_name=FAKE_NETWORK_DHCP_NS,
+                                     callback=mock.ANY)
+        process_instance.enable.assert_called_once()
 
     def test_disable_isolated_metadata_proxy(self):
         method_path = ('neutron.agent.metadata.driver.MetadataDriver'

neutron/tests/unit/agent/metadata/test_driver.py

@@ -18,6 +18,7 @@
 from unittest import mock
 
 from neutron_lib import constants
+from neutron_lib import exceptions as lib_exceptions
 from neutron_lib import fixture as lib_fixtures
 from oslo_config import cfg
 from oslo_utils import uuidutils
@@ -241,6 +242,26 @@ def test_spawn_metadata_proxy(self):
     def test_spawn_metadata_proxy_dad_failed(self):
         self._test_spawn_metadata_proxy(dad_failed=True)
 
+    @mock.patch.object(metadata_driver.LOG, 'error')
+    def test_spawn_metadata_proxy_handles_process_exception(self, error_log):
+        process_instance = mock.Mock(active=False)
+        process_instance.enable.side_effect = (
+            lib_exceptions.ProcessExecutionError('Something happened', -1))
+        with mock.patch.object(metadata_driver.MetadataDriver,
+                               '_get_metadata_proxy_process_manager',
+                               return_value=process_instance):
+            process_monitor = mock.Mock()
+            network_id = 123456
+            metadata_driver.MetadataDriver.spawn_monitored_metadata_proxy(
+                process_monitor,
+                'dummy_namespace',
+                self.METADATA_PORT,
+                cfg.CONF,
+                network_id=network_id)
+        error_log.assert_called_once()
+        process_monitor.register.assert_not_called()
+        self.assertNotIn(network_id, metadata_driver.MetadataDriver.monitors)
+
     def test_create_config_file_wrong_user(self):
         with mock.patch('pwd.getpwnam', side_effect=KeyError):
             config = metadata_driver.HaproxyConfigurator(_uuid(),

neutron/tests/unit/agent/ovn/metadata/test_driver.py

@@ -16,6 +16,7 @@
 import os
 from unittest import mock
 
+from neutron_lib import exceptions as lib_exceptions
 from neutron_lib import fixture as lib_fixtures
 from oslo_config import cfg
 from oslo_utils import uuidutils
@@ -112,6 +113,29 @@ def test_spawn_metadata_proxy(self):
                                           run_as_root=True)
             ])
 
+    @mock.patch.object(metadata_driver.LOG, 'error')
+    def test_spawn_metadata_proxy_handles_process_exception(self, error_log):
+        process_instance = mock.Mock(active=False)
+        process_instance.enable.side_effect = (
+            lib_exceptions.ProcessExecutionError('Something happened', -1))
+
+        with mock.patch.object(metadata_driver.MetadataDriver,
+                               '_get_metadata_proxy_process_manager',
+                               return_value=process_instance):
+            process_monitor = mock.Mock()
+            network_id = 123456
+
+            metadata_driver.MetadataDriver.spawn_monitored_metadata_proxy(
+                process_monitor,
+                'dummy_namespace',
+                self.METADATA_PORT,
+                cfg.CONF,
+                network_id=network_id)
+
+        error_log.assert_called_once()
+        process_monitor.register.assert_not_called()
+        self.assertNotIn(network_id, metadata_driver.MetadataDriver.monitors)
+
     def test_create_config_file_wrong_user(self):
         with mock.patch('pwd.getpwnam', side_effect=KeyError):
             config = metadata_driver.HaproxyConfigurator(mock.ANY, mock.ANY,