Refactor OVN client QoS extension

The QoS OVN client extension is moved to the ML2 driver. This
extension is called from the OVN driver in the events of:
- create port
- update port
- delete port
- update network

The QoS OVN client extension now can accept several rules per policy
as documented in the SUPPORTED_RULES. The QoS OVN client extension
can write one OVN QoS rule per flow direction and each OVN QoS rule
register can hold both a bandwidth limit rule and a DSCP marking rule.

The "update_policy" method is called from the OVN QoS driver, when
a QoS policy or its rules are updated.

The QoS OVN client extension updates the QoS OVN registers
exclusively, based on the related events.

Closes-Bug: #1863852

Change-Id: I4833ed0c9a2741bdd007d4ebb3e8c1cb4c30d4c7

neutron/cmd/ovn/neutron_ovn_db_sync_util.py

@@ -81,8 +81,7 @@ class OVNMechanismDriver(mech_driver.OVNMechanismDriver):
         port['network'] = context.network.current
         # FIXME(lucasagomes): PortContext does not have a session, therefore
         # we need to use the _plugin_context attribute.
-        self.ovn_client.delete_port(context._plugin_context, port['id'],
-                                    port_object=port)
+        self.ovn_client.delete_port(context._plugin_context, port)
 
 
 class AgentNotifierApi(object):

neutron/objects/qos/binding.py

@@ -14,7 +14,10 @@
 #    under the License.
 
 from neutron_lib.objects import common_types
+from sqlalchemy import and_
+from sqlalchemy import exists
 
+from neutron.db import models_v2
 from neutron.db.qos import models as qos_db_model
 from neutron.objects import base
 
@@ -34,6 +37,20 @@ class QosPolicyPortBinding(base.NeutronDbObject):
     primary_keys = ['port_id']
     fields_no_update = ['policy_id', 'port_id']
 
+    @classmethod
+    def get_ports_by_network_id(cls, context, network_id, policy_id=None):
+        query = context.session.query(models_v2.Port).filter(
+            models_v2.Port.network_id == network_id)
+        if policy_id:
+            query = query.filter(exists().where(and_(
+                cls.db_model.port_id == models_v2.Port.id,
+                cls.db_model.policy_id == policy_id)))
+        else:
+            query = query.filter(~exists().where(
+                cls.db_model.port_id == models_v2.Port.id)).filter(
+                models_v2.Port.network_id == network_id)
+        return query.all()
+
 
 @base.NeutronObjectRegistry.register
 class QosPolicyNetworkBinding(base.NeutronDbObject):

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

@@ -30,7 +30,6 @@ from neutron_lib import context as n_context
 from neutron_lib import exceptions as n_exc
 from neutron_lib.plugins import directory
 from neutron_lib.plugins.ml2 import api
-from neutron_lib.services.qos import constants as qos_consts
 from oslo_config import cfg
 from oslo_db import exception as os_db_exc
 from oslo_log import log
@@ -90,9 +89,6 @@ class OVNMechanismDriver(api.MechanismDriver):
     update network/port case, all data validation must be done within
     methods that are part of the database transaction.
     """
-
-    supported_qos_rule_types = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT]
-
     def initialize(self):
         """Perform driver initialization.
 
@@ -118,7 +114,7 @@ class OVNMechanismDriver(api.MechanismDriver):
             LOG.warning('Firewall driver configuration is ignored')
         self._setup_vif_port_bindings()
         self.subscribe()
-        self.qos_driver = qos_driver.OVNQosNotificationDriver.create(self)
+        self.qos_driver = qos_driver.OVNQosDriver.create(self)
         self.trunk_driver = trunk_driver.OVNTrunkDriver.create(self)
 
     @property
@@ -399,8 +395,9 @@ class OVNMechanismDriver(api.MechanismDriver):
         # https://bugs.launchpad.net/neutron/+bug/1739798 is fixed.
         if context._plugin_context.session.is_active:
             return
-        self._ovn_client.update_network(context._plugin_context,
-                                        context.current)
+        self._ovn_client.update_network(
+            context._plugin_context, context.current,
+            original_network=context.original)
 
     def delete_network_postcommit(self, context):
         """Delete a network.

neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/extensions/qos.py

@@ -0,0 +1,251 @@
+# Copyright 2020 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron.objects.qos import binding as qos_binding
+from neutron.objects.qos import policy as qos_policy
+from neutron.objects.qos import rule as qos_rule
+from neutron_lib import constants
+from neutron_lib import context as n_context
+from neutron_lib.plugins import directory
+from neutron_lib.services.qos import constants as qos_consts
+from oslo_log import log as logging
+
+from neutron.common.ovn import utils
+
+
+LOG = logging.getLogger(__name__)
+OVN_QOS_DEFAULT_RULE_PRIORITY = 2002
+
+
+class OVNClientQosExtension(object):
+    """OVN client QoS extension"""
+
+    def __init__(self, driver):
+        LOG.info('Starting OVNClientQosExtension')
+        super(OVNClientQosExtension, self).__init__()
+        self._driver = driver
+        self._plugin_property = None
+
+    @property
+    def _plugin(self):
+        if self._plugin_property is None:
+            self._plugin_property = directory.get_plugin()
+        return self._plugin_property
+
+    @staticmethod
+    def _qos_rules(context, policy_id):
+        """QoS Neutron rules classified per direction and type
+
+        :param context: (context) Neutron request context
+        :param policy_id: (string) Neutron QoS policy ID
+        :return: (dict) nested dictionary of QoS rules, classified per
+                 direction and rule type
+                 {egress: {bw_limit: {max_kbps, max_burst_kbps},
+                           dscp: {dscp_mark}
+                  ingress: {...} }
+        """
+        qos_rules = {constants.EGRESS_DIRECTION: {},
+                     constants.INGRESS_DIRECTION: {}}
+        if policy_id is None:
+            return qos_rules
+
+        # The policy might not have any rule
+        all_rules = qos_rule.get_rules(qos_policy.QosPolicy,
+                                       context, policy_id)
+        for rule in all_rules:
+            if isinstance(rule, qos_rule.QosBandwidthLimitRule):
+                r = {rule.rule_type: {'max_kbps': rule.max_kbps}}
+                if rule.max_burst_kbps:
+                    r[rule.rule_type]['max_burst_kbps'] = rule.max_burst_kbps
+                qos_rules[rule.direction].update(r)
+            elif isinstance(rule, qos_rule.QosDscpMarkingRule):
+                r = {rule.rule_type: {'dscp_mark': rule.dscp_mark}}
+                qos_rules[constants.EGRESS_DIRECTION].update(r)
+            else:
+                LOG.warning('Rule type %(rule_type)s from QoS policy '
+                            '%(policy_id)s is not supported in OVN',
+                            {'rule_type': rule.rule_type,
+                             'policy_id': policy_id})
+        return qos_rules
+
+    def _ovn_qos_rule(self, rules_direction, rules, port_id, network_id,
+                      delete=False):
+        """Generate an OVN QoS register based on several Neutron QoS rules
+
+        A OVN QoS register can contain "bandwidth" and "action" parameters.
+        "bandwidth" defines the rate speed limitation; "action" contains the
+        DSCP value to apply. Both are not exclusive.
+        Only one rule per port and direction can be applied; that's why
+        two rules (bandwidth limit and DSCP) in the same direction must be
+        combined in one OVN QoS register.
+        http://www.openvswitch.org/support/dist-docs/ovn-nb.5.html
+
+        :param rules_direction: (string) rules direction (egress, ingress).
+        :param rules: (dict) {bw_limit: {max_kbps, max_burst_kbps},
+                              dscp: {dscp_mark}}
+        :param port_id: (string) port ID.
+        :param network_id: (string) network ID.
+        :param delete: (bool) defines if this rule if going to be a partial
+                       one (without any bandwidth or DSCP information) to be
+                       used only as deletion rule.
+        :return: (dict) OVN QoS rule register to be used with QoSAddCommand
+                 and QoSDelCommand.
+        """
+        if not delete and not rules:
+            return
+
+        lswitch_name = utils.ovn_name(network_id)
+
+        if rules_direction == constants.EGRESS_DIRECTION:
+            direction = 'from-lport'
+            match = 'inport == "{}"'.format(port_id)
+        else:
+            direction = 'to-lport'
+            match = 'outport == "{}"'.format(port_id)
+
+        ovn_qos_rule = {'switch': lswitch_name, 'direction': direction,
+                        'priority': OVN_QOS_DEFAULT_RULE_PRIORITY,
+                        'match': match}
+
+        if delete:
+            # Any specific rule parameter is left undefined.
+            return ovn_qos_rule
+
+        for rule_type, rule in rules.items():
+            if rule_type == qos_consts.RULE_TYPE_BANDWIDTH_LIMIT:
+                ovn_qos_rule['rate'] = rule['max_kbps']
+                if rule.get('max_burst_kbps'):
+                    ovn_qos_rule['burst'] = rule['max_burst_kbps']
+            elif rule_type == qos_consts.RULE_TYPE_DSCP_MARKING:
+                ovn_qos_rule.update({'dscp': rule['dscp_mark']})
+
+        return ovn_qos_rule
+
+    def _port_effective_qos_policy_id(self, port):
+        """Return port effective QoS policy
+
+        If the port does not have any QoS policy reference or is a network
+        device, then return None.
+        """
+        policy_exists = bool(port.get('qos_policy_id') or
+                             port.get('qos_network_policy_id'))
+        if not policy_exists or utils.is_network_device_port(port):
+            return None, None
+
+        if port.get('qos_policy_id'):
+            return port['qos_policy_id'], 'port'
+        else:
+            return port['qos_network_policy_id'], 'network'
+
+    def _update_port_qos_rules(self, txn, port_id, network_id, qos_policy_id,
+                               qos_rules):
+        # NOTE(ralonsoh): we don't use the transaction context because the
+        # QoS policy could belong to another user (network QoS policy).
+        admin_context = n_context.get_admin_context()
+
+        # Generate generic deletion rules for both directions. In case of
+        # creating deletion rules, the rule content is irrelevant.
+        for ovn_rule in [self._ovn_qos_rule(direction, {}, port_id,
+                                            network_id, delete=True)
+                         for direction in constants.VALID_DIRECTIONS]:
+            txn.add(self._driver._nb_idl.qos_del(**ovn_rule))
+
+        if not qos_policy_id:
+            return  # If no QoS policy is defined, there are no QoS rules.
+
+        # TODO(ralonsoh): for update_network and update_policy operations,
+        # the QoS rules can be retrieved only once.
+        qos_rules = qos_rules or self._qos_rules(admin_context, qos_policy_id)
+        for direction, rules in qos_rules.items():
+            ovn_rule = self._ovn_qos_rule(direction, rules, port_id,
+                                          network_id)
+            if ovn_rule:
+                txn.add(self._driver._nb_idl.qos_add(**ovn_rule))
+
+    def create_port(self, txn, port):
+        self.update_port(txn, port, None, reset=True)
+
+    def delete_port(self, txn, port):
+        self.update_port(txn, port, None, delete=True)
+
+    def update_port(self, txn, port, original_port, reset=False, delete=False,
+                    qos_rules=None):
+        if not reset and not original_port:
+            # If there is no information about the previous QoS policy, do not
+            # make any change, unless the port is new or the QoS information
+            # must be reset (delete any previous configuration and set new
+            # one).
+            return
+
+        qos_policy_id = (None if delete else
+                         self._port_effective_qos_policy_id(port)[0])
+        if not reset and not delete:
+            original_qos_policy_id = self._port_effective_qos_policy_id(
+                original_port)[0]
+            if qos_policy_id == original_qos_policy_id:
+                return  # No QoS policy change
+
+        self._update_port_qos_rules(txn, port['id'], port['network_id'],
+                                    qos_policy_id, qos_rules)
+
+    def update_network(self, txn, network, original_network, reset=False,
+                       qos_rules=None):
+        updated_port_ids = set([])
+        if not reset and not original_network:
+            # If there is no information about the previous QoS policy, do not
+            # make any change.
+            return updated_port_ids
+
+        qos_policy_id = network.get('qos_policy_id')
+        if not reset:
+            original_qos_policy_id = original_network.get('qos_policy_id')
+            if qos_policy_id == original_qos_policy_id:
+                return updated_port_ids  # No QoS policy change
+
+        # NOTE(ralonsoh): we don't use the transaction context because some
+        # ports can belong to other projects.
+        admin_context = n_context.get_admin_context()
+        for port in qos_binding.QosPolicyPortBinding.get_ports_by_network_id(
+                admin_context, network['id']):
+            if utils.is_network_device_port(port):
+                continue
+
+            self._update_port_qos_rules(txn, port['id'], network['id'],
+                                        qos_policy_id, qos_rules)
+            updated_port_ids.add(port['id'])
+
+        return updated_port_ids
+
+    def update_policy(self, context, policy):
+        updated_port_ids = set([])
+        bound_networks = policy.get_bound_networks()
+        bound_ports = policy.get_bound_ports()
+        qos_rules = self._qos_rules(context, policy.id)
+        # TODO(ralonsoh): we need to benchmark this transaction in systems with
+        # a huge amount of ports. This can take a while and could block other
+        # operations.
+        with self._driver._nb_idl.transaction(check_error=True) as txn:
+            for network_id in bound_networks:
+                network = {'qos_policy_id': policy.id, 'id': network_id}
+                updated_port_ids.update(
+                    self.update_network(txn, network, {}, reset=True,
+                                        qos_rules=qos_rules))
+
+            # Update each port bound to this policy, not handled previously in
+            # the network update loop
+            port_ids = [p for p in bound_ports if p not in updated_port_ids]
+            for port in self._plugin.get_ports(context,
+                                               filters={'id': port_ids}):
+                self.update_port(txn, port, {}, reset=True,
+                                 qos_rules=qos_rules)

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

@@ -39,8 +39,10 @@ from neutron.common.ovn import constants as ovn_const
 from neutron.common.ovn import utils
 from neutron.conf.plugins.ml2.drivers.ovn import ovn_conf
 from neutron.db import ovn_revision_numbers_db as db_rev
+from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.extensions \
+    import qos as qos_extension
 from neutron.scheduler import l3_ovn_scheduler
-from neutron.services.qos.drivers.ovn import driver as qos_driver
+
 
 LOG = log.getLogger(__name__)
 
@@ -65,7 +67,8 @@ class OVNClient(object):
         self._plugin_property = None
         self._l3_plugin_property = None
 
-        self._qos_driver = qos_driver.OVNQosDriver(self)
+        # TODO(ralonsoh): handle the OVN client extensions with an ext. manager
+        self._qos_driver = qos_extension.OVNClientQosExtension(self)
         self._ovn_scheduler = l3_ovn_scheduler.get_scheduler()
 
     @property
@@ -431,13 +434,7 @@ class OVNClient(object):
             if self.is_dns_required_for_port(port):
                 self.add_txns_to_sync_port_dns_records(txn, port)
 
-            # Add qos for port by qos table of logical flow instead of tc
-            qos_options = self._qos_driver.get_qos_options(port)
-
-            if qos_options:
-                qos_rule_column = self._create_qos_rules(qos_options,
-                                                         port, lswitch_name)
-                txn.add(self._nb_idl.qos_add(**qos_rule_column))
+            self._qos_driver.create_port(txn, port)
 
         db_rev.bump_revision(context, port, ovn_const.TYPE_PORTS)
 
@@ -466,10 +463,9 @@ class OVNClient(object):
 
     # TODO(lucasagomes): The ``port_object`` parameter was added to
     # keep things backward compatible. Remove it in the Rocky release.
-    def update_port(self, context, port, qos_options=None, port_object=None):
+    def update_port(self, context, port, port_object=None):
         if utils.is_lsp_ignored(port):
             return
-        # Does not need to add qos rule to port_info
         port_info = self._get_port_options(port)
         external_ids = {ovn_const.OVN_PORT_NAME_EXT_ID_KEY: port['name'],
                         ovn_const.OVN_DEVID_EXT_ID_KEY: port['device_id'],
@@ -484,7 +480,6 @@ class OVNClient(object):
                         ovn_const.OVN_REV_NUM_EXT_ID_KEY: str(
                             utils.get_revision_number(
                                 port, ovn_const.TYPE_PORTS))}
-        lswitch_name = utils.ovn_name(port['network_id'])
         sg_cache = {}
         subnet_cache = {}
 
@@ -655,50 +650,7 @@ class OVNClient(object):
                                             addrs_add=addr_add,
                                             addrs_remove=addr_remove))
 
-            # Update QoS policy rule, delete the old one, then add the new one
-            # If we create port with qos_policy, we also need execute
-            # update_port method, which qos_policy is in port dict, so we
-            # also need get policy from port dict if qos_options is None
-            qos_options_new = (qos_options if qos_options
-                               else self._qos_driver.get_qos_options(port))
-            # If port_object is None, we also need to get necessary params
-            # to delete the qos rule
-            qos_options_old = (self._qos_driver.get_qos_options(port_object)
-                               if port_object else qos_options_new)
-
-            ovn_net = self._nb_idl.get_lswitch(lswitch_name)
-            ovn_net_qos_policy = (ovn_net.external_ids
-                                  [ovn_const.OVN_QOS_POLICY_EXT_ID_KEY]
-                                  if ovn_const.OVN_QOS_POLICY_EXT_ID_KEY
-                                  in ovn_net.external_ids else None)
-
-            if qos_options_new:
-                qos_rule_column_old = self._create_qos_rules(qos_options_old,
-                                                             port,
-                                                             lswitch_name,
-                                                             if_delete=True)
-                # Delete old QoS rule first
-                txn.add(self._nb_idl.qos_del(**qos_rule_column_old))
-                # Add new QoS rule
-                qos_rule_column_new = self._create_qos_rules(qos_options_new,
-                                                             port,
-                                                             lswitch_name)
-                txn.add(self._nb_idl.qos_add(**qos_rule_column_new))
-            # If we want to delete port qos_rule by using
-            # param '--no-qos-policy'
-            elif qos_options_old:
-
-                qos_rule_column_old = self._create_qos_rules(qos_options_old,
-                                                             port,
-                                                             lswitch_name,
-                                                             if_delete=True)
-                # Delete old QoS rule
-                txn.add(self._nb_idl.qos_del(**qos_rule_column_old))
-
-            # If we want to delete network qos_rule by using
-            # param '--no-qos-policy'
-            elif not qos_options_old and ovn_net_qos_policy:
-                txn.add(self._nb_idl.qos_del(lswitch_name))
+            self._qos_driver.update_port(txn, port, port_object)
 
             if self.is_dns_required_for_port(port):
                 self.add_txns_to_sync_port_dns_records(
@@ -710,36 +662,6 @@ class OVNClient(object):
         if check_rev_cmd.result == ovn_const.TXN_COMMITTED:
             db_rev.bump_revision(context, port, ovn_const.TYPE_PORTS)
 
-    def _create_qos_rules(self, qos_options, port, lswitch_name,
-                          if_delete=False):
-        qos_rule = {}
-        direction = 'from-lport' if qos_options['direction'] ==\
-                    'egress' else 'to-lport'
-        qos_rule.update(switch=lswitch_name, direction=direction,
-                        priority=2002)
-
-        if direction == 'from-lport':
-            match = 'inport == ' + '"{}"'.format(port['id'])
-            qos_rule.update(match=match)
-        else:
-            match = 'outport == ' + '"{}"'.format(port['id'])
-            qos_rule.update(match=match)
-        # QoS of bandwidth_limit
-        if 'qos_max_rate' in qos_options:
-            burst = qos_options.get('qos_burst')
-            qos_rule.update(rate=qos_options['qos_max_rate'],
-                            burst=burst, dscp=None)
-        # QoS of dscp
-        elif 'dscp_mark' in qos_options:
-            qos_rule.update(rate=None, burst=None,
-                            dscp=qos_options['dscp_mark'])
-        # There is no need 'rate', 'burst' or 'dscp' for deleted method
-        if if_delete is True:
-            qos_rule.pop('rate')
-            qos_rule.pop('burst')
-            qos_rule.pop('dscp')
-        return qos_rule
-
     def _delete_port(self, port_id, port_object=None):
         ovn_port = self._nb_idl.lookup('Logical_Switch_Port', port_id)
         network_id = ovn_port.external_ids.get(
@@ -769,20 +691,10 @@ class OVNClient(object):
                             name=utils.ovn_addrset_name(sg_id, ip_version),
                             addrs_add=None,
                             addrs_remove=addr_list))
-            # Delete qos rule of port
-            try:
-                if (not port_object or 'qos_policy_id' not in port_object or
-                        port_object['qos_policy_id'] is None):
-                    pass
-                else:
-                    qos_options = self._qos_driver.get_qos_options(port_object)
-                    qos_rule_column = self._create_qos_rules(qos_options,
-                                                             port_object,
-                                                             network_id,
-                                                             if_delete=True)
-                    txn.add(self._nb_idl.qos_del(**qos_rule_column))
-            except KeyError:
-                pass
+
+            p_object = ({'id': port_id, 'network_id': network_id}
+                        if not port_object else port_object)
+            self._qos_driver.delete_port(txn, p_object)
 
             if port_object and self.is_dns_required_for_port(port_object):
                 self.add_txns_to_remove_port_dns_records(txn, port_object)
@@ -1751,14 +1663,6 @@ class OVNClient(object):
             ovn_const.OVN_REV_NUM_EXT_ID_KEY: str(
                 utils.get_revision_number(network, ovn_const.TYPE_NETWORKS))}}
 
-        # NOTE(lucasagomes): There's a difference between the
-        # "qos_policy_id" key existing and it being None, the latter is a
-        # valid value. Since we can't save None in OVSDB, we are converting
-        # it to "null" as a placeholder.
-        if 'qos_policy_id' in network:
-            params['external_ids'][ovn_const.OVN_QOS_POLICY_EXT_ID_KEY] = (
-                network['qos_policy_id'] or 'null')
-
         # Enable IGMP snooping if igmp_snooping_enable is enabled in Neutron
         value = 'true' if ovn_conf.is_igmp_snooping_enabled() else 'false'
         params['other_config'] = {ovn_const.MCAST_SNOOP: value,
@@ -1794,21 +1698,6 @@ class OVNClient(object):
         db_rev.delete_revision(
             context, network_id, ovn_const.TYPE_NETWORKS)
 
-    def _is_qos_update_required(self, network):
-        # Is qos service enabled
-        if 'qos_policy_id' not in network:
-            return False
-
-        # Check if qos service wasn't enabled before
-        ovn_net = self._nb_idl.get_lswitch(utils.ovn_name(network['id']))
-        if ovn_const.OVN_QOS_POLICY_EXT_ID_KEY not in ovn_net.external_ids:
-            return True
-
-        # Check if the policy_id has changed
-        new_qos_id = network['qos_policy_id'] or 'null'
-        return new_qos_id != ovn_net.external_ids[
-            ovn_const.OVN_QOS_POLICY_EXT_ID_KEY]
-
     def set_gateway_mtu(self, context, prov_net, txn=None):
         ports = self._plugin.get_ports(
             context, filters=dict(network_id=[prov_net['id']],
@@ -1824,10 +1713,8 @@ class OVNClient(object):
                             name=lrp_name, if_exists=True, options=options))
         self._transaction(commands, txn=txn)
 
-    def update_network(self, context, network):
+    def update_network(self, context, network, original_network=None):
         lswitch_name = utils.ovn_name(network['id'])
-        # Check if QoS needs to be update, before updating OVNDB
-        qos_update_required = self._is_qos_update_required(network)
         check_rev_cmd = self._nb_idl.check_revision_number(
             lswitch_name, network, ovn_const.TYPE_NETWORKS)
 
@@ -1880,9 +1767,9 @@ class OVNClient(object):
                     self.set_gateway_mtu(n_context.get_admin_context(),
                                          network, txn)
 
+            self._qos_driver.update_network(txn, network, original_network)
+
         if check_rev_cmd.result == ovn_const.TXN_COMMITTED:
-            if qos_update_required:
-                self._qos_driver.update_network(network)
             db_rev.bump_revision(context, network, ovn_const.TYPE_NETWORKS)
 
     def _add_subnet_dhcp_options(self, subnet, network,

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

@@ -10,19 +10,14 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from neutron.objects.qos import policy as qos_policy
-from neutron.objects.qos import rule as qos_rule
 from neutron_lib.api.definitions import portbindings
 from neutron_lib import constants
-from neutron_lib import context as n_context
 from neutron_lib.db import constants as db_consts
-from neutron_lib.plugins import directory
 from neutron_lib.services.qos import base
 from neutron_lib.services.qos import constants as qos_consts
 from oslo_config import cfg
 from oslo_log import log as logging
 
-from neutron.common.ovn import utils
 
 LOG = logging.getLogger(__name__)
 
@@ -45,130 +40,24 @@ VIF_TYPES = [portbindings.VIF_TYPE_OVS, portbindings.VIF_TYPE_VHOST_USER]
 VNIC_TYPES = [portbindings.VNIC_NORMAL]
 
 
-class OVNQosNotificationDriver(base.DriverBase):
+class OVNQosDriver(base.DriverBase):
     """OVN notification driver for QoS."""
 
-    def __init__(self, name='OVNQosDriver',
-                 vif_types=VIF_TYPES,
-                 vnic_types=VNIC_TYPES,
-                 supported_rules=SUPPORTED_RULES,
-                 requires_rpc_notifications=False):
-        super(OVNQosNotificationDriver, self).__init__(
-            name, vif_types, vnic_types, supported_rules,
-            requires_rpc_notifications)
-
     @classmethod
     def create(cls, plugin_driver):
-        cls._driver = plugin_driver
-        return cls()
+        obj = OVNQosDriver(name='OVNQosDriver',
+                           vif_types=VIF_TYPES,
+                           vnic_types=VNIC_TYPES,
+                           supported_rules=SUPPORTED_RULES,
+                           requires_rpc_notifications=False)
+        obj._driver = plugin_driver
+        return obj
 
     @property
     def is_loaded(self):
         return OVN_QOS in cfg.CONF.ml2.extension_drivers
 
-    def create_policy(self, context, policy):
-        # No need to update OVN on create
-        pass
-
     def update_policy(self, context, policy):
         # Call into OVN client to update the policy
+        # TODO(ralonsoh): make the same call but to public methods/variables.
         self._driver._ovn_client._qos_driver.update_policy(context, policy)
-
-    def delete_policy(self, context, policy):
-        # No need to update OVN on delete
-        pass
-
-
-class OVNQosDriver(object):
-    """Qos driver for OVN"""
-
-    def __init__(self, driver):
-        LOG.info("Starting OVNQosDriver")
-        super(OVNQosDriver, self).__init__()
-        self._driver = driver
-        self._plugin_property = None
-
-    @property
-    def _plugin(self):
-        if self._plugin_property is None:
-            self._plugin_property = directory.get_plugin()
-        return self._plugin_property
-
-    def _generate_port_options(self, context, policy_id):
-        if policy_id is None:
-            return {}
-        options = {}
-        # The policy might not have any rules
-        all_rules = qos_rule.get_rules(qos_policy.QosPolicy,
-                                       context, policy_id)
-        for rule in all_rules:
-            if isinstance(rule, qos_rule.QosBandwidthLimitRule):
-                options['qos_max_rate'] = rule.max_kbps
-                if rule.max_burst_kbps:
-                    options['qos_burst'] = rule.max_burst_kbps
-                options['direction'] = rule.direction
-            if isinstance(rule, qos_rule.QosDscpMarkingRule):
-                options['dscp_mark'] = rule.dscp_mark
-                options['direction'] = constants.EGRESS_DIRECTION
-        return options
-
-    def get_qos_options(self, port):
-        # Is qos service enabled
-        if 'qos_policy_id' not in port:
-            return {}
-        # Don't apply qos rules to network devices
-        if utils.is_network_device_port(port):
-            return {}
-
-        # Determine if port or network policy should be used
-        context = n_context.get_admin_context()
-        port_policy_id = port.get('qos_policy_id')
-        network_policy_id = None
-        if not port_policy_id:
-            network_policy = qos_policy.QosPolicy.get_network_policy(
-                context, port['network_id'])
-            network_policy_id = network_policy.id if network_policy else None
-
-        # Generate qos options for the selected policy
-        policy_id = port_policy_id or network_policy_id
-        return self._generate_port_options(context, policy_id)
-
-    def _update_network_ports(self, context, network_id, options):
-        # Retrieve all ports for this network
-        ports = self._plugin.get_ports(context,
-                                       filters={'network_id': [network_id]})
-        for port in ports:
-            # Don't apply qos rules if port has a policy
-            port_policy_id = port.get('qos_policy_id')
-            if port_policy_id:
-                continue
-            # Don't apply qos rules to network devices
-            if utils.is_network_device_port(port):
-                continue
-            # Call into OVN client to update port
-            self._driver.update_port(port, qos_options=options)
-
-    def update_network(self, network):
-        # Is qos service enabled
-        if 'qos_policy_id' not in network:
-            return
-
-        # Update the qos options on each network port
-        context = n_context.get_admin_context()
-        options = self._generate_port_options(
-            context, network['qos_policy_id'])
-        self._update_network_ports(context, network.get('id'), options)
-
-    def update_policy(self, context, policy):
-        options = self._generate_port_options(context, policy.id)
-
-        # Update each network bound to this policy
-        network_bindings = policy.get_bound_networks()
-        for network_id in network_bindings:
-            self._update_network_ports(context, network_id, options)
-
-        # Update each port bound to this policy
-        port_bindings = policy.get_bound_ports()
-        for port_id in port_bindings:
-            port = self._plugin.get_port(context, port_id)
-            self._driver.update_port(port, qos_options=options)

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

@@ -0,0 +1,113 @@
+# Copyright 2020 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import mock
+from neutron_lib import constants
+from neutron_lib.services.qos import constants as qos_constants
+
+from neutron.common.ovn import utils as ovn_utils
+from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.extensions \
+    import qos as qos_extension
+from neutron.tests.functional import base
+
+
+QOS_RULE_BW_1 = {'max_kbps': 200, 'max_burst_kbps': 100}
+QOS_RULE_BW_2 = {'max_kbps': 300}
+QOS_RULE_DSCP_1 = {'dscp_mark': 16}
+QOS_RULE_DSCP_2 = {'dscp_mark': 20}
+
+QOS_RULES_1 = {
+    constants.EGRESS_DIRECTION: {
+        qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_1,
+        qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_1},
+    constants.INGRESS_DIRECTION: {
+        qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_2}
+}
+
+QOS_RULES_2 = {
+    constants.EGRESS_DIRECTION: {
+        qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_2,
+        qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_2}
+}
+
+QOS_RULES_3 = {
+    constants.INGRESS_DIRECTION: {
+        qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_1,
+        qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_1}
+}
+
+
+class _OVNClient(object):
+
+    def __init__(self, nd_idl):
+        self._nb_idl = nd_idl
+
+
+class TestOVNClientQosExtension(base.TestOVNFunctionalBase):
+
+    def setUp(self, maintenance_worker=False):
+        super(TestOVNClientQosExtension, self).setUp(
+            maintenance_worker=maintenance_worker)
+        self._add_logical_switch()
+        _ovn_client = _OVNClient(self.nb_api)
+        self.qos_driver = qos_extension.OVNClientQosExtension(_ovn_client)
+
+    def _add_logical_switch(self):
+        self.network_1 = 'network_1'
+        with self.nb_api.transaction(check_error=True) as txn:
+            txn.add(self.nb_api.ls_add(ovn_utils.ovn_name(self.network_1)))
+
+    def _check_rules(self, rules, port_id, network_id):
+        egress_ovn_rule = self.qos_driver._ovn_qos_rule(
+            constants.EGRESS_DIRECTION, rules.get(constants.EGRESS_DIRECTION),
+            port_id, network_id)
+        ingress_ovn_rule = self.qos_driver._ovn_qos_rule(
+            constants.INGRESS_DIRECTION,
+            rules.get(constants.INGRESS_DIRECTION), port_id, network_id)
+
+        with self.nb_api.transaction(check_error=True):
+            ls = self.qos_driver._driver._nb_idl.lookup(
+                'Logical_Switch', ovn_utils.ovn_name(self.network_1))
+            self.assertEqual(len(rules), len(ls.qos_rules))
+            for rule in ls.qos_rules:
+                ref_rule = (egress_ovn_rule if rule.direction == 'from-lport'
+                            else ingress_ovn_rule)
+                action = {}
+                if 'dscp' in ref_rule:
+                    action = {'dscp': ref_rule['dscp']}
+                bandwidth = {}
+                if 'rate' in ref_rule:
+                    bandwidth['rate'] = ref_rule['rate']
+                    if ref_rule.get('burst'):
+                        bandwidth['burst'] = ref_rule['burst']
+                self.assertIn(port_id, rule.match)
+                self.assertEqual(action, rule.action)
+                self.assertEqual(bandwidth, rule.bandwidth)
+
+    def test__update_port_qos_rules(self):
+        port = 'port1'
+
+        def update_and_check(qos_rules):
+            with self.nb_api.transaction(check_error=True) as txn, \
+                    mock.patch.object(self.qos_driver,
+                                      '_qos_rules') as mock_rules:
+                mock_rules.return_value = qos_rules
+                self.qos_driver._update_port_qos_rules(
+                    txn, port, self.network_1, 'qos1', None)
+            self._check_rules(qos_rules, port, self.network_1)
+
+        update_and_check(QOS_RULES_1)
+        update_and_check(QOS_RULES_2)
+        update_and_check(QOS_RULES_3)
+        update_and_check({})

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

@@ -0,0 +1,365 @@
+# Copyright 2020 Red Hat, Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from unittest import mock
+
+import netaddr
+from neutron_lib import constants
+from neutron_lib import context
+from neutron_lib.services.qos import constants as qos_constants
+from oslo_config import cfg
+from oslo_utils import uuidutils
+
+from neutron.core_extensions import qos as core_qos
+from neutron import manager
+from neutron.objects import network as network_obj
+from neutron.objects import ports as port_obj
+from neutron.objects.qos import policy as policy_obj
+from neutron.objects.qos import rule as rule_obj
+from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb.extensions \
+    import qos as qos_extension
+from neutron.tests.unit.plugins.ml2 import test_plugin
+
+
+QOS_RULE_BW_1 = {'max_kbps': 200, 'max_burst_kbps': 100}
+QOS_RULE_BW_2 = {'max_kbps': 300}
+QOS_RULE_DSCP_1 = {'dscp_mark': 16}
+QOS_RULE_DSCP_2 = {'dscp_mark': 20}
+QOS_RULE_MINBW_1 = {'min_kbps': 500}
+
+
+class _Context(object):
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        return
+
+
+class TestOVNClientQosExtension(test_plugin.Ml2PluginV2TestCase):
+
+    CORE_PLUGIN_CLASS = 'neutron.plugins.ml2.plugin.Ml2Plugin'
+    _extension_drivers = ['qos']
+
+    def setUp(self):
+        cfg.CONF.set_override('extension_drivers', self._extension_drivers,
+                              group='ml2')
+        cfg.CONF.set_override('service_plugins', self._extension_drivers)
+        super(TestOVNClientQosExtension, self).setUp()
+        self.setup_coreplugin(self.CORE_PLUGIN_CLASS, load_plugins=True)
+        manager.init()
+        self._mock_qos_loaded = mock.patch.object(
+            core_qos.QosCoreResourceExtension, 'plugin_loaded')
+        self.mock_qos_loaded = self._mock_qos_loaded.start()
+
+        self.txn = _Context()
+        mock_driver = mock.Mock()
+        mock_driver._nb_idl.transaction.return_value = self.txn
+        self.qos_driver = qos_extension.OVNClientQosExtension(mock_driver)
+        self._mock_rules = mock.patch.object(self.qos_driver,
+                                             '_update_port_qos_rules')
+        self.mock_rules = self._mock_rules.start()
+        self.addCleanup(self._mock_rules.stop)
+        self.ctx = context.get_admin_context()
+        self.project_id = uuidutils.generate_uuid()
+        self._initialize_objs()
+
+    def _get_random_db_fields(self, obj_cls=None):
+        obj_cls = obj_cls or self._test_class
+        return obj_cls.modify_fields_to_db(
+            self.get_random_object_fields(obj_cls))
+
+    def _initialize_objs(self):
+        self.qos_policies = []
+        self.ports = []
+        self.networks = []
+        for net_idx in range(2):
+            qos_policy = policy_obj.QosPolicy(
+                self.ctx, id=uuidutils.generate_uuid(),
+                project_id=self.project_id)
+            qos_policy.create()
+            self.qos_policies.append(qos_policy)
+
+            # Any QoS policy should have at least one rule, in order to have
+            # the port dictionary extended with the QoS policy information; see
+            # QoSPlugin._extend_port_resource_request
+            qos_rule = rule_obj.QosDscpMarkingRule(
+                self.ctx, dscp=20, id=uuidutils.generate_uuid(),
+                qos_policy_id=qos_policy.id)
+            qos_rule.create()
+
+            network = network_obj.Network(
+                self.ctx, id=uuidutils.generate_uuid(),
+                project_id=self.project_id)
+            network.create()
+            self.networks.append(network)
+
+            for port_idx in range(3):
+                mac_address = netaddr.EUI(net_idx * 16 + port_idx)
+                port = port_obj.Port(
+                    self.ctx, project_id=self.project_id,
+                    network_id=network.id, device_owner='',
+                    admin_state_up=True, status='DOWN', device_id='2',
+                    mac_address=mac_address)
+                port.create()
+                self.ports.append(port)
+
+    @mock.patch.object(qos_extension.LOG, 'warning')
+    @mock.patch.object(rule_obj, 'get_rules')
+    def test__qos_rules(self, mock_get_rules, mock_warning):
+        rules = [
+            rule_obj.QosBandwidthLimitRule(
+                direction=constants.EGRESS_DIRECTION, **QOS_RULE_BW_1),
+            rule_obj.QosBandwidthLimitRule(
+                direction=constants.INGRESS_DIRECTION, **QOS_RULE_BW_2),
+            rule_obj.QosDscpMarkingRule(**QOS_RULE_DSCP_1),
+            rule_obj.QosMinimumBandwidthRule(**QOS_RULE_MINBW_1)]
+        mock_get_rules.return_value = rules
+        expected = {
+            constants.EGRESS_DIRECTION: {
+                qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_1,
+                qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_1},
+            constants.INGRESS_DIRECTION: {
+                qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_2}
+        }
+        self.assertEqual(expected, self.qos_driver._qos_rules(mock.ANY,
+                                                              'policy_id1'))
+        msg = ('Rule type %(rule_type)s from QoS policy %(policy_id)s is not '
+               'supported in OVN')
+        mock_warning.assert_called_once_with(
+            msg, {'rule_type': qos_constants.RULE_TYPE_MINIMUM_BANDWIDTH,
+                  'policy_id': 'policy_id1'})
+
+    @mock.patch.object(rule_obj, 'get_rules')
+    def test__qos_rules_no_rules(self, mock_get_rules):
+        mock_get_rules.return_value = []
+        expected = {constants.EGRESS_DIRECTION: {},
+                    constants.INGRESS_DIRECTION: {}}
+        self.assertEqual(expected,
+                         self.qos_driver._qos_rules(mock.ANY, mock.ANY))
+
+    def test__ovn_qos_rule_ingress(self):
+        direction = constants.INGRESS_DIRECTION
+        rule = {qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_1}
+        expected = {'burst': 100, 'rate': 200, 'direction': 'to-lport',
+                    'match': 'outport == "port_id"',
+                    'priority': qos_extension.OVN_QOS_DEFAULT_RULE_PRIORITY,
+                    'switch': 'neutron-network_id'}
+        result = self.qos_driver._ovn_qos_rule(
+            direction, rule, 'port_id', 'network_id')
+        self.assertEqual(expected, result)
+
+    def test__ovn_qos_rule_egress(self):
+        direction = constants.EGRESS_DIRECTION
+        rule = {qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_1}
+        expected = {'direction': 'from-lport', 'match': 'inport == "port_id"',
+                    'dscp': 16, 'switch': 'neutron-network_id',
+                    'priority': qos_extension.OVN_QOS_DEFAULT_RULE_PRIORITY}
+        result = self.qos_driver._ovn_qos_rule(
+            direction, rule, 'port_id', 'network_id')
+        self.assertEqual(expected, result)
+
+        rule = {qos_constants.RULE_TYPE_BANDWIDTH_LIMIT: QOS_RULE_BW_2,
+                qos_constants.RULE_TYPE_DSCP_MARKING: QOS_RULE_DSCP_2}
+        expected = {'direction': 'from-lport', 'match': 'inport == "port_id"',
+                    'rate': 300, 'dscp': 20, 'switch': 'neutron-network_id',
+                    'priority': qos_extension.OVN_QOS_DEFAULT_RULE_PRIORITY}
+        result = self.qos_driver._ovn_qos_rule(
+            direction, rule, 'port_id', 'network_id')
+        self.assertEqual(expected, result)
+
+    def test__port_effective_qos_policy_id(self):
+        port = {'qos_policy_id': 'qos1'}
+        self.assertEqual(('qos1', 'port'),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+        port = {'qos_network_policy_id': 'qos1'}
+        self.assertEqual(('qos1', 'network'),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+        port = {'qos_policy_id': 'qos_port',
+                'qos_network_policy_id': 'qos_network'}
+        self.assertEqual(('qos_port', 'port'),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+        port = {}
+        self.assertEqual((None, None),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+        port = {'qos_policy_id': None, 'qos_network_policy_id': None}
+        self.assertEqual((None, None),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+        port = {'qos_policy_id': 'qos1', 'device_owner': 'neutron:port'}
+        self.assertEqual((None, None),
+                         self.qos_driver._port_effective_qos_policy_id(port))
+
+    def test_update_port(self):
+        port = self.ports[0]
+        original_port = self.ports[1]
+
+        # Remove QoS policy
+        original_port.qos_policy_id = self.qos_policies[0].id
+        self.qos_driver.update_port(mock.ANY, port, original_port)
+        self.mock_rules.assert_called_once_with(
+            mock.ANY, port.id, port.network_id, None, None)
+
+        # Change from port policy (qos_policy0) to network policy (qos_policy1)
+        self.mock_rules.reset_mock()
+        port.qos_network_policy_id = self.qos_policies[1].id
+        self.qos_driver.update_port(mock.ANY, port, original_port)
+        self.mock_rules.assert_called_once_with(
+            mock.ANY, port.id, port.network_id, self.qos_policies[1].id, None)
+
+        # No change (qos_policy0)
+        self.mock_rules.reset_mock()
+        port.qos_policy_id = self.qos_policies[0].id
+        original_port.qos_policy_id = self.qos_policies[0].id
+        self.qos_driver.update_port(mock.ANY, port, original_port)
+        self.mock_rules.assert_not_called()
+
+        # No change (no policy)
+        self.mock_rules.reset_mock()
+        port.qos_policy_id = None
+        port.qos_network_policy_id = None
+        original_port.qos_policy_id = None
+        original_port.qos_network_policy_id = None
+        self.qos_driver.update_port(mock.ANY, port, original_port)
+        self.mock_rules.assert_not_called()
+
+        # Reset (no policy)
+        self.qos_driver.update_port(mock.ANY, port, original_port, reset=True)
+        self.mock_rules.assert_called_once_with(
+            mock.ANY, port.id, port.network_id, None, None)
+
+        # Reset (qos_policy0, regardless of being the same a in the previous
+        # state)
+        self.mock_rules.reset_mock()
+        port.qos_policy_id = self.qos_policies[0].id
+        original_port.qos_policy_id = self.qos_policies[1].id
+        self.qos_driver.update_port(mock.ANY, port, original_port, reset=True)
+        self.mock_rules.assert_called_once_with(
+            mock.ANY, port.id, port.network_id, self.qos_policies[0].id, None)
+
+    def test_update_network(self):
+        """Test update network.
+
+        net1: [(1) from qos_policy0 to no QoS policy,
+               (2) from qos_policy0 to qos_policy1]
+        - port10: no QoS port policy
+        - port11: qos_policy0
+        - port12: qos_policy1
+        """
+        policies_ports = [
+            (None, {self.ports[0].id}),
+            (self.qos_policies[1].id, {self.ports[0].id})]
+
+        self.ports[1].qos_policy_id = self.qos_policies[0].id
+        self.ports[1].update()
+        self.ports[2].qos_policy_id = self.qos_policies[1].id
+        self.ports[2].update()
+        for qos_policy_id, reference_ports in policies_ports:
+            self.networks[0].qos_policy_id = qos_policy_id
+            self.networks[0].update()
+            original_network = {'qos_policy_id': self.qos_policies[0]}
+            reviewed_port_ids = self.qos_driver.update_network(
+                mock.ANY, self.networks[0], original_network)
+            self.assertEqual(reference_ports, reviewed_port_ids)
+            calls = [mock.call(mock.ANY, self.ports[0].id,
+                               self.ports[0].network_id, qos_policy_id,
+                               None)]
+            self.mock_rules.assert_has_calls(calls)
+            self.mock_rules.reset_mock()
+
+    def test_update_network_no_policy_change(self):
+        """Test update network if the QoS policy is the same.
+
+        net1: [(1) from qos_policy0 to qos_policy0,
+               (2) from no QoS policy to no QoS policy]
+        """
+        for qos_policy_id in (self.qos_policies[0].id, None):
+            self.networks[0].qos_policy_id = qos_policy_id
+            self.networks[0].update()
+            original_network = {'qos_policy_id': qos_policy_id}
+            reviewed_port_ids = self.qos_driver.update_network(
+                mock.ANY, self.networks[0], original_network)
+            self.assertEqual(set([]), reviewed_port_ids)
+            self.mock_rules.assert_not_called()
+
+    def test_update_network_reset(self):
+        """Test update network.
+
+        net1: [(1) from qos_policy1 to qos_policy1,
+               (2) from no QoS policy to no QoS policy]
+        - port10: no QoS port policy
+        - port11: qos_policy0
+        - port12: qos_policy1
+        """
+        policies_ports = [
+            (self.qos_policies[1].id, {self.ports[0].id}),
+            (None, {self.ports[0].id})]
+
+        self.ports[1].qos_policy_id = self.qos_policies[0].id
+        self.ports[1].update()
+        self.ports[2].qos_policy_id = self.qos_policies[1].id
+        self.ports[2].update()
+        for qos_policy_id, reference_ports in policies_ports:
+            self.networks[0].qos_policy_id = qos_policy_id
+            self.networks[0].update()
+            original_network = {'qos_policy_id': self.qos_policies[0]}
+            reviewed_port_ids = self.qos_driver.update_network(
+                mock.ANY, self.networks[0], original_network, reset=True)
+            self.assertEqual(reference_ports, reviewed_port_ids)
+            calls = [mock.call(mock.ANY, self.ports[0].id,
+                               self.ports[0].network_id, qos_policy_id, None)]
+            self.mock_rules.assert_has_calls(calls)
+            self.mock_rules.reset_mock()
+
+    def test_update_policy(self):
+        """Test update QoS policy, networks and ports bound are updated.
+
+        QoS policy updated: qos_policy0
+        net1: no QoS policy
+        - port10: no port QoS policy
+        - port11: qos_policy0  --> handled during "update_port" and updated
+        - port12: qos_policy1
+        net2: qos_policy0
+        - port20: no port QoS policy  --> handled during "update_network"
+                                          and updated
+        - port21: qos_policy0  --> handled during "update_network", not updated
+                                   handled during "update_port" and updated
+        - port22: qos_policy1  --> handled during "update_network", not updated
+        """
+        self.ports[1].qos_policy_id = self.qos_policies[0].id
+        self.ports[1].update()
+        self.ports[2].qos_policy_id = self.qos_policies[1].id
+        self.ports[2].update()
+        self.ports[4].qos_policy_id = self.qos_policies[0].id
+        self.ports[4].update()
+        self.ports[5].qos_policy_id = self.qos_policies[1].id
+        self.ports[5].update()
+        self.networks[1].qos_policy_id = self.qos_policies[0].id
+        self.networks[1].update()
+        mock_qos_rules = mock.Mock()
+        with mock.patch.object(self.qos_driver, '_qos_rules',
+                               return_value=mock_qos_rules):
+            self.qos_driver.update_policy(self.ctx, self.qos_policies[0])
+        updated_ports = [self.ports[1], self.ports[3], self.ports[4]]
+        calls = [mock.call(self.txn, port.id, port.network_id,
+                           self.qos_policies[0].id, mock_qos_rules)
+                 for port in updated_ports]
+        # We can't ensure the call order because we are not enforcing any order
+        # when retrieving the port and the network list.
+        self.mock_rules.assert_has_calls(calls, any_order=True)

neutron/tests/unit/plugins/ml2/drivers/ovn/mech_driver/test_mech_driver.py

@@ -84,6 +84,7 @@ class TestOVNMechanismDriver(test_plugin.Ml2PluginV2TestCase):
         self.mech_driver = mm.mech_drivers['ovn'].obj
         self.mech_driver._nb_ovn = fakes.FakeOvsdbNbOvnIdl()
         self.mech_driver._sb_ovn = fakes.FakeOvsdbSbOvnIdl()
+        self.mech_driver._ovn_client._qos_driver = mock.Mock()
         self.nb_ovn = self.mech_driver._nb_ovn
         self.sb_ovn = self.mech_driver._sb_ovn
 
@@ -1705,6 +1706,7 @@ class OVNMechanismDriverTestCase(test_plugin.Ml2PluginV2TestCase):
         sb_ovn = fakes.FakeOvsdbSbOvnIdl()
         self.mech_driver._nb_ovn = nb_ovn
         self.mech_driver._sb_ovn = sb_ovn
+        self.mech_driver._ovn_client._qos_driver = mock.Mock()
         self.mech_driver._insert_port_provisioning_block = mock.Mock()
         p = mock.patch.object(ovn_utils, 'get_revision_number', return_value=1)
         p.start()
@@ -1843,6 +1845,7 @@ class TestOVNMechanismDriverSegment(test_segment.HostSegmentMappingTestCase):
         sb_ovn = fakes.FakeOvsdbSbOvnIdl()
         self.mech_driver._nb_ovn = nb_ovn
         self.mech_driver._sb_ovn = sb_ovn
+        self.mech_driver._ovn_client._qos_driver = mock.Mock()
         p = mock.patch.object(ovn_utils, 'get_revision_number', return_value=1)
         p.start()
         self.addCleanup(p.stop)
@@ -2452,6 +2455,7 @@ class TestOVNMechanismDriverSecurityGroup(
         sb_ovn = fakes.FakeOvsdbSbOvnIdl()
         self.mech_driver._nb_ovn = nb_ovn
         self.mech_driver._sb_ovn = sb_ovn
+        self.mech_driver._ovn_client._qos_driver = mock.Mock()
         self.ctx = context.get_admin_context()
         revision_plugin.RevisionPlugin()
 
@@ -2751,6 +2755,7 @@ class TestOVNMechanismDriverMetadataPort(test_plugin.Ml2PluginV2TestCase):
         self.mech_driver = mm.mech_drivers['ovn'].obj
         self.mech_driver._nb_ovn = fakes.FakeOvsdbNbOvnIdl()
         self.mech_driver._sb_ovn = fakes.FakeOvsdbSbOvnIdl()
+        self.mech_driver._ovn_client._qos_driver = mock.Mock()
         self.nb_ovn = self.mech_driver._nb_ovn
         self.sb_ovn = self.mech_driver._sb_ovn
         self.ctx = context.get_admin_context()

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

@@ -12,27 +12,22 @@
 
 import mock
 
-from neutron.objects.qos import policy as qos_policy
-from neutron.objects.qos import rule as qos_rule
 from neutron.tests import base
-from neutron_lib import constants
-from oslo_utils import uuidutils
 
-from neutron.common.ovn import utils
 from neutron.services.qos.drivers.ovn import driver
 
+
 context = 'context'
 
 
-class TestOVNQosNotificationDriver(base.BaseTestCase):
+class TestOVNQosDriver(base.BaseTestCase):
 
     def setUp(self):
-        super(TestOVNQosNotificationDriver, self).setUp()
+        super(TestOVNQosDriver, self).setUp()
         self.mech_driver = mock.Mock()
         self.mech_driver._ovn_client = mock.Mock()
         self.mech_driver._ovn_client._qos_driver = mock.Mock()
-        self.driver = driver.OVNQosNotificationDriver.create(
-            self.mech_driver)
+        self.driver = driver.OVNQosDriver.create(self.mech_driver)
         self.policy = "policy"
 
     def test_create_policy(self):
@@ -49,210 +44,3 @@ class TestOVNQosNotificationDriver(base.BaseTestCase):
         self.driver.delete_policy(context, self.policy)
         self.driver._driver._ovn_client._qos_driver.delete_policy.\
             assert_not_called()
-
-
-class TestOVNQosDriver(base.BaseTestCase):
-
-    def setUp(self):
-        super(TestOVNQosDriver, self).setUp()
-        self.plugin = mock.Mock()
-        self.ovn_client = mock.Mock()
-        self.driver = driver.OVNQosDriver(self.ovn_client)
-        self.driver._plugin_property = self.plugin
-        self.port_id = uuidutils.generate_uuid()
-        self.policy_id = uuidutils.generate_uuid()
-        self.network_id = uuidutils.generate_uuid()
-        self.network_policy_id = uuidutils.generate_uuid()
-        self.policy = self._create_fake_policy()
-        self.port = self._create_fake_port()
-        self.bw_rule = self._create_bw_limit_rule()
-        self.bw_expected = {'qos_max_rate': 1000, 'qos_burst': 100000,
-                            'direction': constants.EGRESS_DIRECTION}
-        self.dscp_rule = self._create_dscp_rule()
-        self.dscp_expected = {'dscp_mark': 16,
-                              'direction': constants.EGRESS_DIRECTION}
-
-    def _create_bw_limit_rule(self):
-        rule_obj = qos_rule.QosBandwidthLimitRule()
-        rule_obj.id = uuidutils.generate_uuid()
-        rule_obj.max_kbps = 1000
-        rule_obj.max_burst_kbps = 100000
-        rule_obj.obj_reset_changes()
-        return rule_obj
-
-    def _create_dscp_rule(self):
-        rule_obj = qos_rule.QosDscpMarkingRule()
-        rule_obj.id = uuidutils.generate_uuid()
-        rule_obj.dscp_mark = 16
-        rule_obj.obj_reset_changes()
-        return rule_obj
-
-    def _create_fake_policy(self):
-        policy_dict = {'id': self.network_policy_id}
-        policy_obj = qos_policy.QosPolicy(context, **policy_dict)
-        policy_obj.obj_reset_changes()
-        return policy_obj
-
-    def _create_fake_port(self):
-        return {'id': self.port_id,
-                'qos_policy_id': self.policy_id,
-                'network_id': self.network_id,
-                'device_owner': 'compute:fake'}
-
-    def _create_fake_network(self):
-        return {'id': self.network_id,
-                'qos_policy_id': self.network_policy_id}
-
-    def test__is_network_device_port(self):
-        self.assertFalse(utils.is_network_device_port(self.port))
-        port = self._create_fake_port()
-        port['device_owner'] = constants.DEVICE_OWNER_DHCP
-        self.assertTrue(utils.is_network_device_port(port))
-        port['device_owner'] = 'neutron:LOADBALANCERV2'
-        self.assertTrue(utils.is_network_device_port(port))
-
-    def _generate_port_options(self, policy_id, return_val, expected_result):
-        with mock.patch.object(qos_rule, 'get_rules',
-                               return_value=return_val) as get_rules:
-            options = self.driver._generate_port_options(context, policy_id)
-            if policy_id:
-                get_rules.assert_called_once_with(qos_policy.QosPolicy,
-                                                  context, policy_id)
-            else:
-                get_rules.assert_not_called()
-            self.assertEqual(expected_result, options)
-
-    def test__generate_port_options_no_policy_id(self):
-        self._generate_port_options(None, [], {})
-
-    def test__generate_port_options_no_rules(self):
-        self._generate_port_options(self.policy_id, [], {})
-
-    def test__generate_port_options_with_bw_rule(self):
-        self._generate_port_options(self.policy_id, [self.bw_rule],
-                                    self.bw_expected)
-
-    def test__generate_port_options_with_dscp_rule(self):
-        self._generate_port_options(self.policy_id, [self.dscp_rule],
-                                    self.dscp_expected)
-
-    def _get_qos_options(self, port, port_policy, network_policy):
-        with mock.patch.object(qos_policy.QosPolicy, 'get_network_policy',
-                               return_value=self.policy) as get_network_policy:
-            with mock.patch.object(self.driver, '_generate_port_options',
-                                   return_value={}) as generate_port_options:
-                options = self.driver.get_qos_options(port)
-                if network_policy:
-                    get_network_policy.\
-                        assert_called_once_with(context, self.network_id)
-                    generate_port_options. \
-                        assert_called_once_with(context,
-                                                self.network_policy_id)
-                elif port_policy:
-                    get_network_policy.assert_not_called()
-                    generate_port_options.\
-                        assert_called_once_with(context, self.policy_id)
-                else:
-                    get_network_policy.assert_not_called()
-                    generate_port_options.assert_not_called()
-                self.assertEqual({}, options)
-
-    def test_get_qos_options_no_qos(self):
-        port = self._create_fake_port()
-        port.pop('qos_policy_id')
-        self._get_qos_options(port, False, False)
-
-    def test_get_qos_options_network_port(self):
-        port = self._create_fake_port()
-        port['device_owner'] = constants.DEVICE_OWNER_DHCP
-        self._get_qos_options(port, False, False)
-
-    @mock.patch('neutron_lib.context.get_admin_context', return_value=context)
-    def test_get_qos_options_port_policy(self, *mocks):
-        self._get_qos_options(self.port, True, False)
-
-    @mock.patch('neutron_lib.context.get_admin_context', return_value=context)
-    def test_get_qos_options_network_policy(self, *mocks):
-        port = self._create_fake_port()
-        port['qos_policy_id'] = None
-        self._get_qos_options(port, False, True)
-
-    def _update_network_ports(self, port, called):
-        with mock.patch.object(self.plugin, 'get_ports',
-                               return_value=[port]) as get_ports:
-            with mock.patch.object(self.ovn_client,
-                                   'update_port') as update_port:
-                self.driver._update_network_ports(
-                    context, self.network_id, {})
-                get_ports.assert_called_once_with(
-                    context, filters={'network_id': [self.network_id]})
-                if called:
-                    update_port.assert_called()
-                else:
-                    update_port.assert_not_called()
-
-    def test__update_network_ports_port_policy(self):
-        self._update_network_ports(self.port, False)
-
-    def test__update_network_ports_network_device(self):
-        port = self._create_fake_port()
-        port['device_owner'] = constants.DEVICE_OWNER_DHCP
-        self._update_network_ports(port, False)
-
-    def test__update_network_ports(self):
-        port = self._create_fake_port()
-        port['qos_policy_id'] = None
-        self._update_network_ports(port, True)
-
-    def _update_network(self, network, called):
-        with mock.patch.object(self.driver, '_generate_port_options',
-                               return_value={}) as generate_port_options:
-            with mock.patch.object(self.driver, '_update_network_ports'
-                                   ) as update_network_ports:
-                self.driver.update_network(network)
-                if called:
-                    generate_port_options.assert_called_once_with(
-                        context, self.network_policy_id)
-                    update_network_ports.assert_called_once_with(
-                        context, self.network_id, {})
-                else:
-                    generate_port_options.assert_not_called()
-                    update_network_ports.assert_not_called()
-
-    @mock.patch('neutron_lib.context.get_admin_context', return_value=context)
-    def test_update_network_no_qos(self, *mocks):
-        network = self._create_fake_network()
-        network.pop('qos_policy_id')
-        self._update_network(network, False)
-
-    @mock.patch('neutron_lib.context.get_admin_context', return_value=context)
-    def test_update_network_policy_change(self, *mocks):
-        network = self._create_fake_network()
-        self._update_network(network, True)
-
-    def test_update_policy(self):
-        with mock.patch.object(self.driver, '_generate_port_options',
-                               return_value={}) as generate_port_options, \
-            mock.patch.object(self.policy, 'get_bound_networks',
-                              return_value=[self.network_id]
-                              ) as get_bound_networks, \
-            mock.patch.object(self.driver, '_update_network_ports'
-                              ) as update_network_ports, \
-            mock.patch.object(self.policy, 'get_bound_ports',
-                              return_value=[self.port_id]
-                              ) as get_bound_ports, \
-            mock.patch.object(self.plugin, 'get_port',
-                              return_value=self.port) as get_port, \
-            mock.patch.object(self.ovn_client, 'update_port',
-                              ) as update_port:
-
-            self.driver.update_policy(context, self.policy)
-
-            generate_port_options.assert_called_once_with(
-                context, self.network_policy_id)
-            get_bound_networks.assert_called_once_with()
-            update_network_ports.assert_called_once_with(
-                context, self.network_id, {})
-            get_bound_ports.assert_called_once_with()
-            get_port.assert_called_once_with(context, self.port_id)
-            update_port.assert_called_once_with(self.port, qos_options={})