Synchronise yoga with upstream

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

@@ -2730,7 +2730,8 @@ def add_txns_to_remove_port_dns_records(self, txn, port):
                 txn.add(self._nb_idl.dns_remove_record(
                         ls_dns_record.uuid, ptr_record, if_exists=True))
 
-    def create_ovn_fair_meter(self, meter_name, from_reload=False, txn=None):
+    def _create_ovn_fair_meter(self, meter_name, from_reload=False, txn=None,
+                               stateless=False):
         """Create row in Meter table with fair attribute set to True.
 
         Create a row in OVN's NB Meter table based on well-known name. This
@@ -2745,21 +2746,35 @@ def create_ovn_fair_meter(self, meter_name, from_reload=False, txn=None):
         """
         meter = self._nb_idl.db_find_rows(
             "Meter", ("name", "=", meter_name)).execute(check_error=True)
-        # The meter is created when a log object is created, not by default.
-        # This condition avoids creating the meter if it wasn't there already
+        # The meters are created when a log object is created, not by default.
+        # This condition avoids creating the meter if it wasn't there already.
         commands = []
         if from_reload and not meter:
             return
+
+        burst_limit = cfg.CONF.network_log.burst_limit
+        rate_limit = cfg.CONF.network_log.rate_limit
+        if stateless:
+            meter_name = meter_name + "_stateless"
+            burst_limit = int(burst_limit / 2)
+            rate_limit = int(rate_limit / 2)
+        # The stateless meter is only created once the stateful meter was
+        # successfully created.
+        # The treatment of limits is not equal for stateful and stateless
+        # traffic at a kernel level according to:
+        # https://bugzilla.redhat.com/show_bug.cgi?id=2212952
+        # The stateless meter is created to adjust this issue.
+        meter = self._nb_idl.db_find_rows(
+            "Meter", ("name", "=", meter_name)).execute(check_error=True)
         if meter:
             meter = meter[0]
             meter_band = self._nb_idl.lookup("Meter_Band",
                                              meter.bands[0].uuid, default=None)
             if meter_band:
                 if all((meter.unit == "pktps",
                         meter.fair[0],
-                        meter_band.rate == cfg.CONF.network_log.rate_limit,
-                        meter_band.burst_size ==
-                        cfg.CONF.network_log.burst_limit)):
+                        meter_band.rate == rate_limit,
+                        meter_band.burst_size == burst_limit)):
                     # Meter (and its meter-band) unchanged: noop.
                     return
             # Re-create meter (and its meter-band) with the new attributes.
@@ -2773,10 +2788,15 @@ def create_ovn_fair_meter(self, meter_name, from_reload=False, txn=None):
         commands.append(self._nb_idl.meter_add(
                         name=meter_name,
                         unit="pktps",
-                        rate=cfg.CONF.network_log.rate_limit,
+                        rate=rate_limit,
                         fair=True,
-                        burst_size=cfg.CONF.network_log.burst_limit,
+                        burst_size=burst_limit,
                         may_exist=False,
                         external_ids={ovn_const.OVN_DEVICE_OWNER_EXT_ID_KEY:
                                       log_const.LOGGING_PLUGIN}))
         self._transaction(commands, txn=txn)
+
+    def create_ovn_fair_meter(self, meter_name, from_reload=False, txn=None):
+        self._create_ovn_fair_meter(meter_name, from_reload, txn)
+        self._create_ovn_fair_meter(meter_name, from_reload, txn,
+                                    stateless=True)

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

@@ -424,11 +424,13 @@ def __init__(self, driver):
         self.l3_plugin = directory.get_plugin(constants.L3)
         table = 'Port_Binding'
         events = (self.ROW_UPDATE,)
-        super(PortBindingChassisEvent, self).__init__(
-            events, table, (('type', '=', ovn_const.OVN_CHASSIS_REDIRECT),))
+        super().__init__(events, table, None)
         self.event_name = 'PortBindingChassisEvent'
 
     def match_fn(self, event, row, old):
+        if row.type != ovn_const.OVN_CHASSIS_REDIRECT:
+            return False
+
         if len(old._data) == 1 and 'external_ids' in old._data:
             # NOTE: since [1], the NB logical_router_port.external_ids are
             # copied into the SB port_binding.external_ids. If only the

neutron/services/logapi/drivers/ovn/driver.py

@@ -28,6 +28,7 @@
 from neutron.common.ovn import constants as ovn_const
 from neutron.common.ovn import utils
 from neutron.conf.services import logging as log_cfg
+from neutron.objects import securitygroup as sg_obj
 from neutron.services.logapi.common import db_api
 from neutron.services.logapi.common import sg_callback
 from neutron.services.logapi.drivers import base
@@ -152,9 +153,21 @@ def _remove_acls_log(self, pgs, ovn_txn, log_name=None):
             msg += " for network log {}".format(log_name)
         LOG.info(msg, acl_changes, acl_absents, acl_visits)
 
-    def _set_acls_log(self, pgs, ovn_txn, actions_enabled, log_name):
+    def _set_acls_log(self, pgs, context, ovn_txn, actions_enabled, log_name):
         acl_changes, acl_visits = 0, 0
         for pg in pgs:
+            meter_name = self.meter_name
+            if pg["name"] != ovn_const.OVN_DROP_PORT_GROUP_NAME:
+                sg = sg_obj.SecurityGroup.get_sg_by_id(
+                    context,
+                    pg["external_ids"][ovn_const.OVN_SG_EXT_ID_KEY])
+                if not sg:
+                    LOG.warning("Port Group %s is missing a corresponding "
+                                "security group, skipping its network log "
+                                "setting...", pg["name"])
+                    continue
+                if not sg.stateful:
+                    meter_name = meter_name + ("_stateless")
             for acl_uuid in pg["acls"]:
                 acl_visits += 1
                 acl = self.ovn_nb.lookup("ACL", acl_uuid)
@@ -163,7 +176,7 @@ def _set_acls_log(self, pgs, ovn_txn, actions_enabled, log_name):
                     continue
                 columns = {
                     'log': acl.action in actions_enabled,
-                    'meter': self.meter_name,
+                    'meter': meter_name,
                     'name': log_name,
                     'severity': "info"
                 }
@@ -183,12 +196,13 @@ def _update_log_objs(self, context, ovn_txn, log_objs):
         for log_obj in log_objs:
             pgs = self._pgs_from_log_obj(context, log_obj)
             actions_enabled = self._acl_actions_enabled(log_obj)
-            self._set_acls_log(pgs, ovn_txn, actions_enabled,
+            self._set_acls_log(pgs, context, ovn_txn, actions_enabled,
                                utils.ovn_name(log_obj.id))
 
     def _pgs_all(self):
         return self.ovn_nb.db_list(
-            "Port_Group", columns=["name", "acls"]).execute(check_error=True)
+            "Port_Group",
+            columns=["name", "external_ids", "acls"]).execute(check_error=True)
 
     def _pgs_from_log_obj(self, context, log_obj):
         """Map Neutron log_obj into affected port groups in OVN.
@@ -207,11 +221,13 @@ def _pgs_from_log_obj(self, context, log_obj):
                 # No sg, no port, DROP: return DROP pg
                 if log_obj.event == log_const.DROP_EVENT:
                     return [{"name": pg_drop.name,
-                        "acls": [r.uuid for r in pg_drop.acls]}]
+                             "external_ids": pg_drop.external_ids,
+                             "acls": [r.uuid for r in pg_drop.acls]}]
                 # No sg, no port, ACCEPT: return all except DROP pg
                 pgs = self._pgs_all()
                 pgs.remove({"name": pg_drop.name,
-                    "acls": [r.uuid for r in pg_drop.acls]})
+                            "external_ids": pg_drop.external_ids,
+                            "acls": [r.uuid for r in pg_drop.acls]})
                 return pgs
             except idlutils.RowNotFound:
                 pass
@@ -223,6 +239,7 @@ def _pgs_from_log_obj(self, context, log_obj):
                 pg = self.ovn_nb.lookup("Port_Group",
                                         ovn_const.OVN_DROP_PORT_GROUP_NAME)
                 pgs.append({"name": pg.name,
+                            "external_ids": pg.external_ids,
                             "acls": [r.uuid for r in pg.acls]})
             except idlutils.RowNotFound:
                 pass
@@ -235,6 +252,7 @@ def _pgs_from_log_obj(self, context, log_obj):
                                         utils.ovn_port_group_name(
                                             log_obj.resource_id))
                 pgs.append({"name": pg.name,
+                            "external_ids": pg.external_ids,
                             "acls": [r.uuid for r in pg.acls]})
             except idlutils.RowNotFound:
                 pass
@@ -248,6 +266,7 @@ def _pgs_from_log_obj(self, context, log_obj):
                     pg = self.ovn_nb.lookup("Port_Group",
                                             utils.ovn_port_group_name(sg_id))
                     pgs.append({"name": pg.name,
+                                "external_ids": pg.external_ids,
                                 "acls": [r.uuid for r in pg.acls]})
                 except idlutils.RowNotFound:
                     pass
@@ -266,7 +285,7 @@ def create_log(self, context, log_obj):
         with self.ovn_nb.transaction(check_error=True) as ovn_txn:
             self._ovn_client.create_ovn_fair_meter(self.meter_name,
                                                    txn=ovn_txn)
-            self._set_acls_log(pgs, ovn_txn, actions_enabled,
+            self._set_acls_log(pgs, context, ovn_txn, actions_enabled,
                                utils.ovn_name(log_obj.id))
 
     def create_log_precommit(self, context, log_obj):
@@ -334,7 +353,7 @@ def update_log(self, context, log_obj):
             if not self._unset_disabled_acls(context, log_obj, ovn_txn):
                 pgs = self._pgs_from_log_obj(context, log_obj)
                 actions_enabled = self._acl_actions_enabled(log_obj)
-                self._set_acls_log(pgs, ovn_txn, actions_enabled,
+                self._set_acls_log(pgs, context, ovn_txn, actions_enabled,
                                    utils.ovn_name(log_obj.id))
 
     def delete_log(self, context, log_obj):
@@ -356,6 +375,8 @@ def delete_log(self, context, log_obj):
                 self._remove_acls_log(pgs, ovn_txn)
                 ovn_txn.add(self.ovn_nb.meter_del(self.meter_name,
                                                   if_exists=True))
+                ovn_txn.add(self.ovn_nb.meter_del(
+                    self.meter_name + "_stateless", if_exists=True))
             LOG.info("All ACL logs cleared after deletion of log_obj %s",
                      log_obj.id)
             return

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

@@ -1125,10 +1125,9 @@ def test_check_for_logging_conf_change(self):
         # Check a meter and fair meter exist
         self.assertTrue(self.nb_api._tables['Meter'].rows)
         self.assertTrue(self.nb_api._tables['Meter_Band'].rows)
-        self.assertEqual(cfg.CONF.network_log.burst_limit,
-            [*self.nb_api._tables['Meter_Band'].rows.values()][0].burst_size)
-        self.assertEqual(cfg.CONF.network_log.rate_limit,
-            [*self.nb_api._tables['Meter_Band'].rows.values()][0].rate)
+        self.assertEqual(len([*self.nb_api._tables['Meter'].rows.values()]),
+            len([*self.nb_api._tables['Meter_Band'].rows.values()]))
+        self._check_meters_consistency()
         # Update burst and rate limit values on the configuration
         ovn_config.cfg.CONF.set_override('burst_limit', CFG_NEW_BURST,
                                          group='network_log')
@@ -1138,7 +1137,16 @@ def test_check_for_logging_conf_change(self):
         self.assertRaises(periodics.NeverAgain,
                           self.maint.check_fair_meter_consistency)
         # Check meter band was effectively changed after the maintenance call
-        self.assertEqual(CFG_NEW_BURST,
-            [*self.nb_api._tables['Meter_Band'].rows.values()][0].burst_size)
-        self.assertEqual(CFG_NEW_RATE,
-            [*self.nb_api._tables['Meter_Band'].rows.values()][0].rate)
+        self._check_meters_consistency(CFG_NEW_BURST, CFG_NEW_RATE)
+
+    def _check_meters_consistency(self, new_burst=None, new_rate=None):
+        burst, rate = (new_burst, new_rate) if new_burst else (
+            cfg.CONF.network_log.burst_limit, cfg.CONF.network_log.rate_limit)
+        for meter in [*self.nb_api._tables['Meter'].rows.values()]:
+            meter_band = self.nb_api.lookup('Meter_Band', meter.bands[0].uuid)
+            if "_stateless" in meter.name:
+                self.assertEqual(int(burst / 2), meter_band.burst_size)
+                self.assertEqual(int(rate / 2), meter_band.rate)
+            else:
+                self.assertEqual(burst, meter_band.burst_size)
+                self.assertEqual(rate, meter_band.rate)

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

@@ -577,3 +577,45 @@ def test_ovsdb_probe_interval(self):
         interval = ovn_conf.get_ovn_ovsdb_probe_interval()
         for idl in idls:
             self.assertEqual(interval, idl._session.reconnect.probe_interval)
+
+
+class TestPortBindingChassisEvent(base.TestOVNFunctionalBase,
+                                  test_l3.L3NatTestCaseMixin):
+
+    def setUp(self, **kwargs):
+        super().setUp(**kwargs)
+        self.chassis = self.add_fake_chassis('ovs-host1')
+        self.l3_plugin = directory.get_plugin(plugin_constants.L3)
+        kwargs = {'arg_list': (external_net.EXTERNAL,),
+                  external_net.EXTERNAL: True}
+        self.net = self._make_network(
+            self.fmt, 'ext_net', True, as_admin=True, **kwargs)
+        self._make_subnet(self.fmt, self.net, '20.0.10.1', '20.0.10.0/24')
+        port_res = self._create_port(self.fmt, self.net['network']['id'])
+        self.port = self.deserialize(self.fmt, port_res)['port']
+
+        self.ext_api = test_extensions.setup_extensions_middleware(
+            test_l3.L3TestExtensionManager())
+        self.pb_event_match = mock.patch.object(
+            self.sb_api.idl._portbinding_event, 'match_fn').start()
+
+    def _check_pb_type(self, _type):
+        def check_pb_type(_type):
+            if len(self.pb_event_match.call_args_list) < 1:
+                return False
+
+            pb_row = self.pb_event_match.call_args_list[0].args[1]
+            return _type == pb_row.type
+
+        n_utils.wait_until_true(lambda: check_pb_type(_type), timeout=5)
+
+    def test_pb_type_patch(self):
+        router = self._make_router(self.fmt, self._tenant_id)
+        self._add_external_gateway_to_router(router['router']['id'],
+                                             self.net['network']['id'])
+        self._check_pb_type('patch')
+
+    def test_pb_type_empty(self):
+        self.sb_api.lsp_bind(self.port['id'], self.chassis,
+                             may_exist=True).execute(check_error=True)
+        self._check_pb_type('')

neutron/tests/functional/services/logapi/drivers/ovn/test_driver.py

@@ -30,6 +30,12 @@ def setUp(self):
         self._check_is_supported()
         self.ctxt = context.Context('admin', 'fake_tenant')
 
+        # Since these tests use the _create_network() from the unit test suite
+        # but _create_security_group() is from the functional tests, two
+        # different tenant_ids will be used unless we specify the following
+        # line in the code:
+        self._tenant_id = self.ctxt.project_id
+
     def _check_is_supported(self):
         if not self.log_driver.network_logging_supported(self.nb_api):
             self.skipTest("The current OVN version does not offer support "

neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/ovsdb/test_ovn_client.py

@@ -247,7 +247,16 @@ def test_create_ovn_fair_meter(self):
         self.ovn_client.create_ovn_fair_meter(self._log_driver.meter_name)
         self.assertFalse(self.nb_idl.meter_del.called)
         self.assertTrue(self.nb_idl.meter_add.called)
-        self.nb_idl.meter_add.assert_called_once_with(
+        self.nb_idl.meter_add.assert_any_call(
+            name=self._log_driver.meter_name + "_stateless",
+            unit="pktps",
+            rate=int(self.fake_cfg_network_log.rate_limit / 2),
+            fair=True,
+            burst_size=int(self.fake_cfg_network_log.burst_limit / 2),
+            may_exist=False,
+            external_ids={constants.OVN_DEVICE_OWNER_EXT_ID_KEY:
+                          log_const.LOGGING_PLUGIN})
+        self.nb_idl.meter_add.assert_any_call(
             name=self._log_driver.meter_name,
             unit="pktps",
             rate=self.fake_cfg_network_log.rate_limit,
@@ -259,10 +268,17 @@ def test_create_ovn_fair_meter(self):
 
     def test_create_ovn_fair_meter_unchanged(self):
         mock_find_rows = mock.Mock()
-        mock_find_rows.execute.return_value = [self._fake_meter()]
+        fake_meter1 = [self._fake_meter()]
+        fake_meter2 = [self._fake_meter(
+            name=self._log_driver.meter_name + "_stateless",
+            bands=[mock.Mock(uuid='tb_stateless')])]
+        mock_find_rows.execute.side_effect = [fake_meter1, fake_meter1,
+                                              fake_meter2, fake_meter2]
         self.nb_idl.db_find_rows.return_value = mock_find_rows
         self.nb_idl.lookup.side_effect = lambda table, key, default: (
-            self._fake_meter_band() if key == "test_band" else default)
+            self._fake_meter_band() if key == "test_band" else
+            self._fake_meter_band_stateless() if key == "tb_stateless" else
+            default)
         self.ovn_client.create_ovn_fair_meter(self._log_driver.meter_name)
         self.assertFalse(self.nb_idl.meter_del.called)
         self.assertFalse(self.nb_idl.meter_add.called)