[OVN] Fix rate and burst for stateless security groups

Right now, as per kernel limitation, the burst limit is not correctly
enforcing the rate and burst when using the ovn "log-related" option and
stateless security groups. We log exactly double the burst. Creating a
new meter that limits the rate and burst to half of the expected ones is
a workaround that solves the issue.

Closes-bug: #2032929

Conflicts:
 - neutron/services/logapi/drivers/ovn/driver.py
 - neutron/tests/functional/services/logapi/drivers/ovn/test_driver.py

Signed-off-by: Elvira García <[email protected]>
Change-Id: Ib0047d38c58bcebb23c8887e7934987ff8c8a432
(cherry picked from commit a3a113a)

Author: elvgarrui

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/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,17 @@ 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 ovn_const.OVN_DROP_PORT_GROUP_NAME not in pg["name"]:
+                stateful = (sg_obj.SecurityGroup
+                            .get_sg_by_id(context, pg["name"]
+                                          .replace('pg_', '', 1)
+                                          .replace('_', '-')).stateful)
+                if not 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 +172,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,7 +192,7 @@ 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):
@@ -266,7 +275,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 +343,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 +365,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/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)

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

@@ -21,6 +21,7 @@
 
 from neutron.common.ovn import constants as ovn_const
 from neutron.common.ovn import utils as ovn_utils
+from neutron.objects import securitygroup as sg_obj
 from neutron.services.logapi.drivers.ovn import driver as ovn_driver
 from neutron.tests import base
 from neutron.tests.unit import fake_resources
@@ -92,6 +93,15 @@ def _fake_meter_band(self, **kwargs):
         meter_band_obj_dict = {**meter_band_defaults_dict, **kwargs}
         return mock.Mock(**meter_band_obj_dict)
 
+    def _fake_meter_band_stateless(self, **kwargs):
+        meter_band_defaults_dict = {
+            'uuid': 'tb_stateless',
+            'rate': int(self.fake_cfg_network_log.rate_limit / 2),
+            'burst_size': int(self.fake_cfg_network_log.burst_limit / 2),
+        }
+        meter_band_obj_dict = {**meter_band_defaults_dict, **kwargs}
+        return mock.Mock(**meter_band_obj_dict)
+
 
 class TestOVNDriver(TestOVNDriverBase):
     def test_create(self):
@@ -287,7 +297,8 @@ def _mock_lookup(_pg_table, acl_uuid, default):
                          self._nb_ovn.db_set.call_count)
 
     @mock.patch.object(ovn_driver.LOG, 'info')
-    def test__set_acls_log(self, m_info):
+    @mock.patch.object(sg_obj.SecurityGroup, 'get_sg_by_id')
+    def test__set_acls_log(self, get_sg, m_info):
         pg_dict = self._fake_pg_dict(acls=['acl1', 'acl2', 'acl3', 'acl4'])
         log_name = 'test_obj_name'
         used_name = 'test_used_name'
@@ -297,10 +308,14 @@ def _mock_lookup(_pg_table, acl_uuid):
                 return self._fake_acl()
             return self._fake_acl(name=used_name)
 
+        sg = fake_resources.FakeSecurityGroup.create_one_security_group(
+            attrs={'stateful': True})
+        get_sg.return_value = sg
         self._nb_ovn.lookup.side_effect = _mock_lookup
         actions_enabled = self._log_driver._acl_actions_enabled(
             self._fake_log_obj(event=log_const.ALL_EVENT))
-        self._log_driver._set_acls_log([pg_dict], self._nb_ovn.transaction,
+        self._log_driver._set_acls_log([pg_dict], self.context,
+                                       self._nb_ovn.transaction,
                                        actions_enabled, log_name)
         info_args, _info_kwargs = m_info.call_args_list[0]
         self.assertIn('Set %d (out of %d visited) ACLs for network log %s',