Fix ingress bandwidth limit in the openvswitch agent

For ingress bandwidth limiting openvswitch agent is using QoS and queues
from the Open vSwitch. There is always queue 0 used for that purpose.
Initially, when this feature was implemented, we assumed that queue 0 is
kind of the "default" queue to which all traffic will be send if there
are no other queues. But that's not true thus ingress bandwidth limiting
wasn't working properly with this agent.

This patch fixes that issue but adding in the table=0 of the br-int
additional OF rule to send all traffic to the queue 0.
In this queue for some ports there can be QoS configured
and then it will be applied for the port. If port don't have any QoS
configured, nothing will happen and all will work like before this
patch.

Biggest problem with that solution was the case when also ports with
minimum bandwidth are on the same node becuase such ports are using
different queues (queue number is the same as ofport number of the tap
interface).
In case when traffic is going from the port with minimum bandwidth QoS
to the port which has ingress bw limit configured, traffic is going only
through br-int and will use queue 0 to apply ingress bw limit properly.
In case when traffic from port with minimum bandwidth set needs to go
out from the host, it will always use physical bridge (minimum bandwidth
is only supported for the provider networks) and proper queue will be
set for such traffic in the physical bridge.
To be able to set proper queue in the physical bridge, this patch adds
additional OF rule to the br-int to set queue_num value in the pkt_mark
field [1] as this seems to be only field which can "survive" passing
bridges.

[1] https://man7.org/linux/man-pages/man7/ovs-fields.7.html

Closes-Bug: #1959567
Change-Id: I1e31565475f38c6ad817268699b165759ac05410
(cherry picked from commit f7ab90baad)
(cherry picked from commit 5573230d30)

neutron/agent/common/ovs_lib.py

@@ -707,6 +707,17 @@ class OVSBridge(BaseOVS):
         LOG.info("Port %(port_id)s not present in bridge %(br_name)s",
                  {'port_id': port_id, 'br_name': self.br_name})
 
+    def get_bridge_patch_ports_ofports(self):
+        ports = self.ovsdb.db_find(
+            'Interface', ('type', '=', 'patch'),
+            columns=['name', 'ofport']).execute()
+        patch_ports = []
+        for port in ports:
+            if self.br_name != self.get_bridge_for_iface(port['name']):
+                continue
+            patch_ports.append(port['ofport'])
+        return patch_ports
+
     def delete_ports(self, all_ports=False):
         if all_ports:
             port_names = self.get_port_name_list()
@@ -783,6 +794,23 @@ class OVSBridge(BaseOVS):
                     other_config=other_config))
         return qos_uuid
 
+    def set_queue_for_ingress_bandwidth_limit(self):
+        # reg3 is used to memoize if queue was set or not. If it is first visit
+        # to table 0 for a packet (i.e. reg3 == 0), set queue and memoize (i.e.
+        # load 1 to reg3), then goto table 0 again. The packet will be handled
+        # as usual when the second visit to table 0.
+        # For min bw reg4 is used for the same purpose. In case if there we
+        # would need one of those registries for something else in the future
+        # we can try to use same reg4 for both OF rules, this one and the one
+        # which sets pkt_mark for minimum bandwidth and play with bitmask
+        self.add_flow(
+            table=constants.LOCAL_SWITCHING,
+            reg3=0,
+            priority=200,
+            actions=("set_queue:%s,load:1->NXM_NX_REG3[0],"
+                     "resubmit(,%s)" % (QOS_DEFAULT_QUEUE,
+                                        constants.LOCAL_SWITCHING)))
+
     def _update_ingress_bw_limit_for_port(
             self, port_name, max_kbps, max_burst_kbps):
         queue_id = self._update_queue(
@@ -909,7 +937,7 @@ class OVSBridge(BaseOVS):
         min_bps = queue['other_config'].get('min-rate')
         return int(int(min_bps) / 1000) if min_bps else None
 
-    def _set_queue_for_minimum_bandwidth(self, queue_num):
+    def _set_pkt_mark_for_minimum_bandwidth(self, queue_num):
         # reg4 is used to memoize if queue was set or not. If it is first visit
         # to table 0 for a packet (i.e. reg4 == 0), set queue and memoize (i.e.
         # load 1 to reg4), then goto table 0 again. The packet will be handled
@@ -919,10 +947,26 @@ class OVSBridge(BaseOVS):
             in_port=queue_num,
             reg4=0,
             priority=200,
-            actions=("set_queue:%s,load:1->NXM_NX_REG4[0],"
+            actions=("set_field:%s->pkt_mark,load:1->NXM_NX_REG4[0],"
                      "resubmit(,%s)" % (queue_num, constants.LOCAL_SWITCHING)))
 
-    def _unset_queue_for_minimum_bandwidth(self, queue_num):
+    def set_queue_for_minimum_bandwidth(self, queue_num):
+        # reg4 is used to memoize if queue was set or not. If it is first visit
+        # to table 0 for a packet (i.e. reg4 == 0), set queue and memoize (i.e.
+        # load 1 to reg4), then goto table 0 again. The packet will be handled
+        # as usual when the second visit to table 0.
+        patch_ports = self.get_bridge_patch_ports_ofports()
+        for patch_port in patch_ports:
+            self.add_flow(
+                table=0,
+                in_port=patch_port,
+                pkt_mark=queue_num,
+                reg4=0,
+                priority=200,
+                actions=("set_queue:%s,load:1->NXM_NX_REG4[0],"
+                         "resubmit(,0)" % queue_num))
+
+    def _unset_pkt_mark_for_minimum_bandwidth(self, queue_num):
         self.delete_flows(
             table=constants.LOCAL_SWITCHING,
             in_port=queue_num,
@@ -947,7 +991,7 @@ class OVSBridge(BaseOVS):
             qos_id=qos_id, queues=qos_queues)
         for egress_port_name in egress_port_names:
             self._set_port_qos(egress_port_name, qos_id=qos_id)
-        self._set_queue_for_minimum_bandwidth(queue_num)
+        self._set_pkt_mark_for_minimum_bandwidth(queue_num)
         return qos_id
 
     def delete_minimum_bandwidth_queue(self, port_id):
@@ -955,7 +999,7 @@ class OVSBridge(BaseOVS):
         if not queue:
             return
         queue_num = int(queue['external_ids']['queue-num'])
-        self._unset_queue_for_minimum_bandwidth(queue_num)
+        self._unset_pkt_mark_for_minimum_bandwidth(queue_num)
         qos_id, qos_queues = self._find_qos(
             self._min_bw_qos_id,
             qos_constants.RULE_TYPE_MINIMUM_BANDWIDTH)
@@ -963,10 +1007,15 @@ class OVSBridge(BaseOVS):
             return
         if queue_num in qos_queues.keys():
             qos_queues.pop(queue_num)
-            self._update_qos(
-                self._min_bw_qos_id,
-                qos_constants.RULE_TYPE_MINIMUM_BANDWIDTH,
-                qos_id=qos_id, queues=qos_queues)
+            if qos_queues:
+                self._update_qos(
+                    self._min_bw_qos_id,
+                    qos_constants.RULE_TYPE_MINIMUM_BANDWIDTH,
+                    qos_id=qos_id, queues=qos_queues)
+            self.ovsdb.db_clear('Port', port_id, 'qos').execute(
+                check_error=False)
+            if not qos_queues:
+                self._delete_qos(qos_id)
             self._delete_queue(
                 queue['_uuid'], qos_constants.RULE_TYPE_MINIMUM_BANDWIDTH)
 

neutron/plugins/ml2/drivers/openvswitch/agent/extension_drivers/qos_driver.py

@@ -50,6 +50,7 @@ class QosOVSAgentDriver(qos.QosLinuxAgentDriver):
         be down until the QoS rules are rebuilt.
         """
         self.br_int.clear_bandwidth_qos()
+        self.br_int.set_queue_for_ingress_bandwidth_limit()
 
     def initialize(self):
         self.br_int = self.agent_api.request_int_br()
@@ -188,6 +189,8 @@ class QosOVSAgentDriver(qos.QosLinuxAgentDriver):
             egress_port_names.extend(ports)
         qos_id = self.br_int.update_minimum_bandwidth_queue(
             port['port_id'], egress_port_names, vif_port.ofport, rule.min_kbps)
+        for phy_br in self.agent_api.request_phy_brs():
+            phy_br.set_queue_for_minimum_bandwidth(vif_port.ofport)
         LOG.debug('Minimum bandwidth egress rule was updated/created for port '
                   '%(port_id)s and rule %(rule_id)s. QoS ID: %(qos_id)s. '
                   'Egress ports with QoS applied: %(ports)s',

neutron/tests/fullstack/test_qos.py

@@ -28,10 +28,13 @@ from neutron.tests.fullstack.resources import environment
 from neutron.tests.fullstack.resources import machine
 from neutron.tests.unit import testlib_api
 
+from neutron.agent.common import ovs_lib
 from neutron.conf.plugins.ml2.drivers import linuxbridge as \
     linuxbridge_agent_config
 from neutron.plugins.ml2.drivers.linuxbridge.agent import \
     linuxbridge_neutron_agent as linuxbridge_agent
+from neutron.plugins.ml2.drivers.openvswitch.agent.common \
+    import constants as ovs_constants
 from neutron.services.qos.drivers.linuxbridge import driver as lb_drv
 from neutron.services.qos.drivers.openvswitch import driver as ovs_drv
 
@@ -365,6 +368,16 @@ class TestBwLimitQoSOvs(_TestBwLimitQoS, base.BaseFullStackTestCase):
                 lambda: vm.bridge.get_ingress_bw_limit_for_port(
                     vm.port.name) == (limit, burst),
                 timeout=10)
+            br_int_flows = vm.bridge.dump_flows_for_table(
+                ovs_constants.LOCAL_SWITCHING)
+            expected = (
+                'priority=200,reg3=0 '
+                'actions=set_queue:%(queue_num)s,'
+                'load:0x1->NXM_NX_REG3[0],resubmit(,0)' % {
+                    'queue_num': ovs_lib.QOS_DEFAULT_QUEUE
+                }
+            )
+            self.assertIn(expected, br_int_flows)
 
     def test_bw_limit_qos_port_removed(self):
         """Test if rate limit config is properly removed when whole port is

neutron/tests/functional/agent/common/test_ovs_lib.py

@@ -360,6 +360,19 @@ class BaseOVSTestCase(base.BaseSudoTestCase):
         self.assertIsNone(max_rate)
         self.assertIsNone(burst)
 
+    def test_set_pkt_mark_for_ingress_bandwidth_limit(self):
+        self._create_bridge()
+        self.ovs.set_queue_for_ingress_bandwidth_limit()
+        flows = self.ovs.dump_flows_for_table(ovs_constants.LOCAL_SWITCHING)
+        expected = (
+            'priority=200,reg3=0 '
+            'actions=set_queue:%(queue_num)s,'
+            'load:0x1->NXM_NX_REG3[0],resubmit(,0)' % {
+                'queue_num': ovs_lib.QOS_DEFAULT_QUEUE
+            }
+        )
+        self.assertIn(expected, flows)
+
     def test_ingress_bw_limit(self):
         port_name = ('port-' + uuidutils.generate_uuid())[:8]
         self._create_bridge()
@@ -416,18 +429,22 @@ class BaseOVSTestCase(base.BaseSudoTestCase):
         qos = self._list_qos(qos_id)
         self.assertEqual(0, len(qos['queues']))
 
-    def test__set_queue_for_minimum_bandwidth(self):
+    def test__set_pkt_mark_for_minimum_bandwidth(self):
         self._create_bridge()
-        self.ovs._set_queue_for_minimum_bandwidth(1234)
+        self.ovs._set_pkt_mark_for_minimum_bandwidth(1234)
         flows = self.ovs.dump_flows_for_table(ovs_constants.LOCAL_SWITCHING)
-        expected = 'priority=200,reg4=0,in_port=1234 actions=set_queue:1234,' \
-                   'load:0x1->NXM_NX_REG4[0],resubmit(,0)'
+        # NOTE(slaweq) 1234 in dec is 0x4d2,
+        # action set_field:1234->pkt mark is shown in the OF output as
+        # load:0x4d2->NXM_NX_PKT_MARK[]
+        expected = ('priority=200,reg4=0,in_port=1234 '
+                    'actions=load:0x4d2->NXM_NX_PKT_MARK[],'
+                    'load:0x1->NXM_NX_REG4[0],resubmit(,0)')
         self.assertIn(expected, flows)
 
-    def test__unset_queue_for_minimum_bandwidth(self):
-        self.test__set_queue_for_minimum_bandwidth()
+    def test__unset_pkt_mark_for_minimum_bandwidth(self):
+        self.test__set_pkt_mark_for_minimum_bandwidth()
 
-        self.ovs._unset_queue_for_minimum_bandwidth(1234)
+        self.ovs._unset_pkt_mark_for_minimum_bandwidth(1234)
         flows = self.ovs.dump_flows_for_table(ovs_constants.LOCAL_SWITCHING)
         expected = 'in_port=1234'
         self.assertNotIn(expected, flows)
@@ -499,10 +516,7 @@ class BaseOVSTestCase(base.BaseSudoTestCase):
         self.assertEqual(queue_id_1, qos['queues'][1].uuid)
 
         self.ovs.delete_minimum_bandwidth_queue(neutron_port_id_1)
-        self._check_value({'_uuid': qos_id}, self._list_qos, qos_id,
-                          keys_to_check=['_uuid'])
-        qos = self._list_qos(qos_id)
-        self.assertEqual(0, len(qos['queues']))
+        self.assertIsNone(self._list_qos(qos_id))
 
     def test_delete_minimum_bandwidth_queue_no_qos_found(self):
         queue_id, neutron_port_id = self._create_queue(queue_num=1)