[OVN] Import OVN Client, L3 and QoS related code

This patch moves the OVN Client, L3 and QoS related code:
Previous paths in networking-ovn tree:
./networking_ovn/ml2/qos_driver.py -> neutron/services/qos/drivers/ovn/qos_driver.py
./networking_ovn/common/ovn_client.py -> neutron/plugins/ml2/drivers/ovn/mech_driver/ovsdb/ovn_client.py
./networking_ovn/l3/l3_ovn_scheduler.py -> neutron/scheduler/l3_ovn_scheduler.py
./networking_ovn/l3/l3_ovn.py -> neutron/services/ovn_l3/plugin.py

Co-Authored-By: Amitabha Biswas <abiswas@us.ibm.com>
Co-Authored-By: Andrew Austin <aaustin@redhat.com>
Co-Authored-By: Anh Tran <anhtt@vn.fujitsu.com>
Co-Authored-By: Armando Migliaccio <armamig@gmail.com>
Co-Authored-By: Arslan Qadeer <arslanq@xgrid.co>
Co-Authored-By: Boden R <bodenvmw@gmail.com>
Co-Authored-By: Brian Haley <bhaley@redhat.com>
Co-Authored-By: Cao Xuan Hoang <hoangcx@vn.fujitsu.com>
Co-Authored-By: Chandra S Vejendla <csvejend@us.ibm.com>
Co-Authored-By: Changxun Zhou <zhoucx@dtdream.com>
Co-Authored-By: Daniel Alvarez <dalvarez@redhat.com>
Co-Authored-By: Dong Jun <dongj@dtdream.com>
Co-Authored-By: Doug Hellmann <doug@doughellmann.com>
Co-Authored-By: Gary Kotton <gkotton@vmware.com>
Co-Authored-By: Guoshuai Li <ligs@dtdream.com>
Co-Authored-By: Han Zhou <zhouhan@gmail.com>
Co-Authored-By: Ihar Hrachyshka <ihrachys@redhat.com>
Co-Authored-By: Jakub Libosvar <libosvar@redhat.com>
Co-Authored-By: John Kasperski <jckasper@us.ibm.com>
Co-Authored-By: Kamil Sambor <ksambor@redhat.com>
Co-Authored-By: Kevin Benton <kevin@benton.pub>
Co-Authored-By: Kyle Mestery <mestery@mestery.com>
Co-Authored-By: LIU Yulong <liuyulong@le.com>
Co-Authored-By: Lucas Alvares Gomes <lucasagomes@gmail.com>
Co-Authored-By: Maciej Józefczyk <mjozefcz@redhat.com>
Co-Authored-By: Miguel Angel Ajo <majopela@redhat.com>
Co-Authored-By: Na <nazhu@cn.ibm.com>
Co-Authored-By: Numan Siddique <nusiddiq@redhat.com>
Co-Authored-By: Richard Theis <rtheis@us.ibm.com>
Co-Authored-By: Russell Bryant <rbryant@redhat.com>
Co-Authored-By: Sławek Kapłoński <slawek@kaplonski.pl>
Co-Authored-By: Terry Wilson <twilson@redhat.com>
Co-Authored-By: Yunxiang Tao <taoyunxiang@cmss.chinamobile.com>
Co-Authored-By: lzklibj <lzklibj@cn.ibm.com>
Co-Authored-By: melissaml <ma.lei@99cloud.net>
Co-Authored-By: reedip <rbanerje@redhat.com>
Co-Authored-By: venkata anil <anilvenkata@redhat.com>
Co-Authored-By: xurong00037997 <xu.rong@zte.com.cn>
Co-Authored-By: zhufl <zhu.fanglei@zte.com.cn>

Change-Id: I52bc2785d815b3f04efbda3b78a28861ca9e8fe1
Related-Blueprint: neutron-ovn-merge

neutron/db/ovn_revision_numbers_db.py

@@ -23,6 +23,7 @@ from oslo_utils import timeutils
 import sqlalchemy as sa
 from sqlalchemy.orm import exc
 
+from neutron.common.ovn import utils as ovn_utils
 from neutron.db.models import l3  # noqa
 from neutron.db.models import ovn as ovn_models
 from neutron.db.models import securitygroup  # noqa
@@ -87,13 +88,6 @@ class UnknownResourceType(n_exc.NeutronException):
     message = 'Uknown resource type: %(resource_type)s'
 
 
-def get_revision_number(resource, resource_type):
-    """Get the resource's revision number based on its type."""
-    if resource_type in _TYPES_PRIORITY_ORDER:
-        return resource['revision_number']
-    raise UnknownResourceType(resource_type=resource_type)
-
-
 def _get_standard_attr_id(context, resource_uuid, resource_type):
     try:
         row = context.session.query(STD_ATTR_MAP[resource_type]).filter_by(
@@ -110,20 +104,19 @@ def create_initial_revision(context, resource_uuid, resource_type,
     LOG.debug('create_initial_revision uuid=%s, type=%s, rev=%s',
               resource_uuid, resource_type, revision_number)
     db_func = context.session.merge if may_exist else context.session.add
-    with db_api.CONTEXT_WRITER.using(context):
+    with context.session.begin(subtransactions=True):
         std_attr_id = _get_standard_attr_id(
             context, resource_uuid, resource_type)
         row = ovn_models.OVNRevisionNumbers(
             resource_uuid=resource_uuid, resource_type=resource_type,
             standard_attr_id=std_attr_id, revision_number=revision_number)
         db_func(row)
-        context.session.flush()
 
 
 @db_api.retry_if_session_inactive()
 def delete_revision(context, resource_uuid, resource_type):
     LOG.debug('delete_revision(%s)', resource_uuid)
-    with db_api.CONTEXT_WRITER.using(context):
+    with context.session.begin(subtransactions=True):
         row = context.session.query(ovn_models.OVNRevisionNumbers).filter_by(
             resource_uuid=resource_uuid,
             resource_type=resource_type).one_or_none()
@@ -143,7 +136,7 @@ def _ensure_revision_row_exist(context, resource, resource_type):
     # deal with objects that already existed before the sync work. I believe
     # that we can remove this method after few development cycles. Or,
     # if we decide to make a migration script as well.
-    with db_api.CONTEXT_WRITER.using(context):
+    with context.session.begin(subtransactions=True):
         if not context.session.query(ovn_models.OVNRevisionNumbers).filter_by(
                 resource_uuid=resource['id'],
                 resource_type=resource_type).one_or_none():
@@ -158,7 +151,7 @@ def _ensure_revision_row_exist(context, resource, resource_type):
 @db_api.retry_if_session_inactive()
 def get_revision_row(context, resource_uuid):
     try:
-        with db_api.CONTEXT_READER.using(context):
+        with context.session.begin(subtransactions=True):
             return context.session.query(
                 ovn_models.OVNRevisionNumbers).filter_by(
                 resource_uuid=resource_uuid).one()
@@ -168,8 +161,8 @@ def get_revision_row(context, resource_uuid):
 
 @db_api.retry_if_session_inactive()
 def bump_revision(context, resource, resource_type):
-    revision_number = get_revision_number(resource, resource_type)
-    with db_api.CONTEXT_WRITER.using(context):
+    revision_number = ovn_utils.get_revision_number(resource, resource_type)
+    with context.session.begin(subtransactions=True):
         _ensure_revision_row_exist(context, resource, resource_type)
         std_attr_id = _get_standard_attr_id(
             context, resource['id'], resource_type)
@@ -202,7 +195,7 @@ def get_inconsistent_resources(context):
                          whens=MAINTENANCE_CREATE_UPDATE_TYPE_ORDER)
     time_ = (timeutils.utcnow() -
              datetime.timedelta(seconds=INCONSISTENCIES_OLDER_THAN))
-    with db_api.CONTEXT_READER.using(context):
+    with context.session.begin(subtransactions=True):
         query = context.session.query(ovn_models.OVNRevisionNumbers).join(
             standard_attr.StandardAttribute,
             ovn_models.OVNRevisionNumbers.standard_attr_id ==
@@ -231,6 +224,6 @@ def get_deleted_resources(context):
     """
     sort_order = sa.case(value=ovn_models.OVNRevisionNumbers.resource_type,
                          whens=MAINTENANCE_DELETE_TYPE_ORDER)
-    with db_api.CONTEXT_READER.using(context):
+    with context.session.begin(subtransactions=True):
         return context.session.query(ovn_models.OVNRevisionNumbers).filter_by(
             standard_attr_id=None).order_by(sort_order).all()

neutron/scheduler/l3_ovn_scheduler.py

@@ -0,0 +1,154 @@
+#
+#    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 abc
+import copy
+import random
+
+from oslo_log import log
+import six
+
+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
+
+
+LOG = log.getLogger(__name__)
+
+OVN_SCHEDULER_CHANCE = 'chance'
+OVN_SCHEDULER_LEAST_LOADED = 'leastloaded'
+
+
+@six.add_metaclass(abc.ABCMeta)
+class OVNGatewayScheduler(object):
+
+    def __init__(self):
+        pass
+
+    @abc.abstractmethod
+    def select(self, nb_idl, sb_idl, gateway_name, candidates=None):
+        """Schedule the gateway port of a router to an OVN chassis.
+
+        Schedule the gateway router port only if it is not already
+        scheduled.
+        """
+
+    def filter_existing_chassis(self, nb_idl, gw_chassis,
+                                physnet, chassis_physnets,
+                                existing_chassis):
+        chassis_list = copy.copy(existing_chassis)
+        for chassis_name in existing_chassis:
+            if utils.is_gateway_chassis_invalid(chassis_name, gw_chassis,
+                                                physnet, chassis_physnets):
+                LOG.debug("Chassis %(chassis)s is invalid for scheduling "
+                          "router in physnet: %(physnet)s.",
+                          {'chassis': chassis_name,
+                           'physnet': physnet})
+                chassis_list.remove(chassis_name)
+        return chassis_list
+
+    def _schedule_gateway(self, nb_idl, sb_idl, gateway_name, candidates,
+                          existing_chassis):
+        existing_chassis = existing_chassis or []
+        candidates = candidates or self._get_chassis_candidates(sb_idl)
+        candidates = list(set(candidates) - set(existing_chassis))
+        # If no candidates, or gateway scheduled on MAX_GATEWAY_CHASSIS nodes
+        # or all candidates in existing_chassis, return existing_chassis.
+        # Otherwise, if more candidates present, then schedule them.
+        if existing_chassis:
+            if not candidates or (
+                    len(existing_chassis) == ovn_const.MAX_GW_CHASSIS):
+                return existing_chassis
+        if not candidates:
+            return [ovn_const.OVN_GATEWAY_INVALID_CHASSIS]
+        chassis_count = ovn_const.MAX_GW_CHASSIS - len(existing_chassis)
+        # The actual binding of the gateway to a chassis via the options
+        # column or gateway_chassis column in the OVN_Northbound is done
+        # by the caller
+        chassis = self._select_gateway_chassis(
+            nb_idl, candidates)[:chassis_count]
+        # priority of existing chassis is higher than candidates
+        chassis = existing_chassis + chassis
+
+        LOG.debug("Gateway %s scheduled on chassis %s",
+                  gateway_name, chassis)
+        return chassis
+
+    @abc.abstractmethod
+    def _select_gateway_chassis(self, nb_idl, candidates):
+        """Choose a chassis from candidates based on a specific policy."""
+
+    def _get_chassis_candidates(self, sb_idl):
+        # TODO(azbiswas): Allow selection of a specific type of chassis when
+        # the upstream code merges.
+        # return (sb_idl.get_all_chassis('gateway_router') or
+        #    sb_idl.get_all_chassis())
+        return sb_idl.get_all_chassis()
+
+
+class OVNGatewayChanceScheduler(OVNGatewayScheduler):
+    """Randomly select an chassis for a gateway port of a router"""
+
+    def select(self, nb_idl, sb_idl, gateway_name, candidates=None,
+               existing_chassis=None):
+        return self._schedule_gateway(nb_idl, sb_idl, gateway_name,
+                                      candidates, existing_chassis)
+
+    def _select_gateway_chassis(self, nb_idl, candidates):
+        candidates = copy.deepcopy(candidates)
+        random.shuffle(candidates)
+        return candidates
+
+
+class OVNGatewayLeastLoadedScheduler(OVNGatewayScheduler):
+    """Select the least loaded chassis for a gateway port of a router"""
+
+    def select(self, nb_idl, sb_idl, gateway_name, candidates=None,
+               existing_chassis=None):
+        return self._schedule_gateway(nb_idl, sb_idl, gateway_name,
+                                      candidates, existing_chassis)
+
+    @staticmethod
+    def _get_chassis_load_by_prios(chassis_info):
+        """Retrieve the amount of ports by priorities hosted in the chassis.
+
+        @param   chassis_info: list of (port, prio) hosted by this chassis
+        @type    chassis_info: []
+        @return: A list of (prio, number_of_ports) tuples.
+        """
+        chassis_load = {}
+        for lrp, prio in chassis_info:
+            chassis_load[prio] = chassis_load.get(prio, 0) + 1
+        return chassis_load.items()
+
+    @staticmethod
+    def _get_chassis_load(chassis):
+        chassis_ports_prios = chassis[1]
+        return sorted(
+            OVNGatewayLeastLoadedScheduler._get_chassis_load_by_prios(
+                chassis_ports_prios), reverse=True)
+
+    def _select_gateway_chassis(self, nb_idl, candidates):
+        chassis_bindings = nb_idl.get_all_chassis_gateway_bindings(candidates)
+        return [chassis for chassis, load in sorted(chassis_bindings.items(),
+                key=OVNGatewayLeastLoadedScheduler._get_chassis_load)]
+
+
+OVN_SCHEDULER_STR_TO_CLASS = {
+    OVN_SCHEDULER_CHANCE: OVNGatewayChanceScheduler,
+    OVN_SCHEDULER_LEAST_LOADED: OVNGatewayLeastLoadedScheduler}
+
+
+def get_scheduler():
+    return OVN_SCHEDULER_STR_TO_CLASS[ovn_conf.get_ovn_l3_scheduler()]()

neutron/services/ovn_l3/plugin.py

@@ -0,0 +1,403 @@
+#
+#    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.db import dns_db
+from neutron.db import extraroute_db
+from neutron.db import l3_gwmode_db
+from neutron.db.models import l3 as l3_models
+from neutron.quota import resource_registry
+from neutron_lib.api.definitions import external_net
+from neutron_lib.api.definitions import portbindings
+from neutron_lib.api.definitions import provider_net as pnet
+from neutron_lib.callbacks import events
+from neutron_lib.callbacks import registry
+from neutron_lib.callbacks import resources
+from neutron_lib import constants as n_const
+from neutron_lib import context as n_context
+from neutron_lib import exceptions as n_exc
+from neutron_lib.plugins import constants as plugin_constants
+from neutron_lib.plugins import directory
+from neutron_lib.services import base as service_base
+from oslo_log import log
+from oslo_utils import excutils
+
+from neutron.common.ovn import constants as ovn_const
+from neutron.common.ovn import extensions
+from neutron.common.ovn import utils
+from neutron.db import ovn_revision_numbers_db as db_rev
+from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import impl_idl_ovn
+from neutron.plugins.ml2.drivers.ovn.mech_driver.ovsdb import ovn_client
+from neutron.scheduler import l3_ovn_scheduler
+
+
+LOG = log.getLogger(__name__)
+
+
+@registry.has_registry_receivers
+class OVNL3RouterPlugin(service_base.ServicePluginBase,
+                        extraroute_db.ExtraRoute_dbonly_mixin,
+                        l3_gwmode_db.L3_NAT_db_mixin,
+                        dns_db.DNSDbMixin):
+    """Implementation of the OVN L3 Router Service Plugin.
+
+    This class implements a L3 service plugin that provides
+    router and floatingip resources and manages associated
+    request/response.
+    """
+
+    # TODO(mjozefcz): Start consuming it from neutron-lib
+    # once available.
+    supported_extension_aliases = (
+        extensions.ML2_SUPPORTED_API_EXTENSIONS_OVN_L3)
+
+    @resource_registry.tracked_resources(router=l3_models.Router,
+                                         floatingip=l3_models.FloatingIP)
+    def __init__(self):
+        LOG.info("Starting OVNL3RouterPlugin")
+        super(OVNL3RouterPlugin, self).__init__()
+        self._nb_ovn_idl = None
+        self._sb_ovn_idl = None
+        self._plugin_property = None
+        self._ovn_client_inst = None
+        self.scheduler = l3_ovn_scheduler.get_scheduler()
+        self._register_precommit_callbacks()
+
+    def _register_precommit_callbacks(self):
+        registry.subscribe(
+            self.create_router_precommit, resources.ROUTER,
+            events.PRECOMMIT_CREATE)
+        registry.subscribe(
+            self.create_floatingip_precommit, resources.FLOATING_IP,
+            events.PRECOMMIT_CREATE)
+
+    @property
+    def _ovn_client(self):
+        if self._ovn_client_inst is None:
+            self._ovn_client_inst = ovn_client.OVNClient(self._ovn,
+                                                         self._sb_ovn)
+        return self._ovn_client_inst
+
+    @property
+    def _ovn(self):
+        if self._nb_ovn_idl is None:
+            LOG.info("Getting OvsdbNbOvnIdl")
+            conn = impl_idl_ovn.get_connection(impl_idl_ovn.OvsdbNbOvnIdl)
+            self._nb_ovn_idl = impl_idl_ovn.OvsdbNbOvnIdl(conn)
+        return self._nb_ovn_idl
+
+    @property
+    def _sb_ovn(self):
+        if self._sb_ovn_idl is None:
+            LOG.info("Getting OvsdbSbOvnIdl")
+            conn = impl_idl_ovn.get_connection(impl_idl_ovn.OvsdbSbOvnIdl)
+            self._sb_ovn_idl = impl_idl_ovn.OvsdbSbOvnIdl(conn)
+        return self._sb_ovn_idl
+
+    @property
+    def _plugin(self):
+        if self._plugin_property is None:
+            self._plugin_property = directory.get_plugin()
+        return self._plugin_property
+
+    def get_plugin_type(self):
+        return plugin_constants.L3
+
+    def get_plugin_description(self):
+        """returns string description of the plugin."""
+        return ("L3 Router Service Plugin for basic L3 forwarding"
+                " using OVN")
+
+    def create_router_precommit(self, resource, event, trigger, context,
+                                router, router_id, router_db):
+        db_rev.create_initial_revision(
+            context, router_id, ovn_const.TYPE_ROUTERS)
+
+    def create_router(self, context, router):
+        router = super(OVNL3RouterPlugin, self).create_router(context, router)
+        try:
+            self._ovn_client.create_router(router)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                # Delete the logical router
+                LOG.error('Unable to create lrouter for %s', router['id'])
+                super(OVNL3RouterPlugin, self).delete_router(context,
+                                                             router['id'])
+        return router
+
+    def update_router(self, context, id, router):
+        original_router = self.get_router(context, id)
+        result = super(OVNL3RouterPlugin, self).update_router(context, id,
+                                                              router)
+        try:
+            self._ovn_client.update_router(result, original_router)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                LOG.exception('Unable to update lrouter for %s', id)
+                revert_router = {'router': original_router}
+                super(OVNL3RouterPlugin, self).update_router(context, id,
+                                                             revert_router)
+        return result
+
+    def delete_router(self, context, id):
+        original_router = self.get_router(context, id)
+        super(OVNL3RouterPlugin, self).delete_router(context, id)
+        try:
+            self._ovn_client.delete_router(context, id)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                super(OVNL3RouterPlugin, self).create_router(
+                    context, {'router': original_router})
+
+    def _add_neutron_router_interface(self, context, router_id,
+                                      interface_info, may_exist=False):
+        try:
+            router_interface_info = (
+                super(OVNL3RouterPlugin, self).add_router_interface(
+                    context, router_id, interface_info))
+        except n_exc.PortInUse:
+            if not may_exist:
+                raise
+            # NOTE(lucasagomes): If the port is already being used it means
+            # the interface has been created already, let's just fetch it from
+            # the database. Perhaps the code below should live in Neutron
+            # itself, a get_router_interface() method in the main class
+            # would be handy
+            port = self._plugin.get_port(context, interface_info['port_id'])
+            subnets = [self._plugin.get_subnet(context, s)
+                       for s in utils.get_port_subnet_ids(port)]
+            router_interface_info = (
+                self._make_router_interface_info(
+                    router_id, port['tenant_id'], port['id'],
+                    port['network_id'], subnets[0]['id'],
+                    [subnet['id'] for subnet in subnets]))
+
+        return router_interface_info
+
+    def add_router_interface(self, context, router_id, interface_info,
+                             may_exist=False):
+        router_interface_info = self._add_neutron_router_interface(
+            context, router_id, interface_info, may_exist=may_exist)
+        try:
+            self._ovn_client.create_router_port(
+                router_id, router_interface_info)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                super(OVNL3RouterPlugin, self).remove_router_interface(
+                    context, router_id, router_interface_info)
+
+        return router_interface_info
+
+    def remove_router_interface(self, context, router_id, interface_info):
+        router_interface_info = (
+            super(OVNL3RouterPlugin, self).remove_router_interface(
+                context, router_id, interface_info))
+        try:
+            port_id = router_interface_info['port_id']
+            subnet_ids = router_interface_info.get('subnet_ids')
+            self._ovn_client.delete_router_port(
+                context, port_id, router_id=router_id, subnet_ids=subnet_ids)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                super(OVNL3RouterPlugin, self).add_router_interface(
+                    context, router_id, interface_info)
+        return router_interface_info
+
+    def create_floatingip_precommit(self, resource, event, trigger, context,
+                                    floatingip, floatingip_id, floatingip_db):
+        db_rev.create_initial_revision(
+            context, floatingip_id, ovn_const.TYPE_FLOATINGIPS)
+
+    def create_floatingip(self, context, floatingip,
+                          initial_status=n_const.FLOATINGIP_STATUS_DOWN):
+        fip = super(OVNL3RouterPlugin, self).create_floatingip(
+            context, floatingip, initial_status)
+        self._ovn_client.create_floatingip(context, fip)
+        return fip
+
+    def delete_floatingip(self, context, id):
+        # TODO(lucasagomes): Passing ``original_fip`` object as a
+        # parameter to the OVNClient's delete_floatingip() method is done
+        # for backward-compatible reasons. Remove it in the Rocky release
+        # of OpenStack.
+        original_fip = self.get_floatingip(context, id)
+        super(OVNL3RouterPlugin, self).delete_floatingip(context, id)
+        self._ovn_client.delete_floatingip(context, id,
+                                           fip_object=original_fip)
+
+    def update_floatingip(self, context, id, floatingip):
+        # TODO(lucasagomes): Passing ``original_fip`` object as a
+        # parameter to the OVNClient's update_floatingip() method is done
+        # for backward-compatible reasons. Remove it in the Rocky release
+        # of OpenStack.
+        original_fip = self.get_floatingip(context, id)
+        fip = super(OVNL3RouterPlugin, self).update_floatingip(context, id,
+                                                               floatingip)
+        self._ovn_client.update_floatingip(context, fip,
+                                           fip_object=original_fip)
+        return fip
+
+    def update_floatingip_status(self, context, floatingip_id, status):
+        fip = super(OVNL3RouterPlugin, self).update_floatingip_status(
+            context, floatingip_id, status)
+        self._ovn_client.update_floatingip_status(context, fip)
+        return fip
+
+    def disassociate_floatingips(self, context, port_id, do_notify=True):
+        fips = self.get_floatingips(context.elevated(),
+                                    filters={'port_id': [port_id]})
+        router_ids = super(OVNL3RouterPlugin, self).disassociate_floatingips(
+            context, port_id, do_notify)
+        for fip in fips:
+            router_id = fip.get('router_id')
+            fixed_ip_address = fip.get('fixed_ip_address')
+            if router_id and fixed_ip_address:
+                update_fip = {'logical_ip': fixed_ip_address,
+                              'external_ip': fip['floating_ip_address']}
+                try:
+                    self._ovn_client.disassociate_floatingip(update_fip,
+                                                             router_id)
+                    self.update_floatingip_status(
+                        context, fip['id'], n_const.FLOATINGIP_STATUS_DOWN)
+                except Exception as e:
+                    LOG.error('Error in disassociating floatingip %(id)s: '
+                              '%(error)s', {'id': fip['id'], 'error': e})
+        return router_ids
+
+    def _get_gateway_port_physnet_mapping(self):
+        # This function returns all gateway ports with corresponding
+        # external network's physnet
+        net_physnet_dict = {}
+        port_physnet_dict = {}
+        l3plugin = directory.get_plugin(plugin_constants.L3)
+        if not l3plugin:
+            return port_physnet_dict
+        context = n_context.get_admin_context()
+        for net in l3plugin._plugin.get_networks(
+                context, {external_net.EXTERNAL: [True]}):
+            if net.get(pnet.NETWORK_TYPE) in [n_const.TYPE_FLAT,
+                                              n_const.TYPE_VLAN]:
+                net_physnet_dict[net['id']] = net.get(pnet.PHYSICAL_NETWORK)
+        for port in l3plugin._plugin.get_ports(context, filters={
+                'device_owner': [n_const.DEVICE_OWNER_ROUTER_GW]}):
+            port_physnet_dict[port['id']] = net_physnet_dict.get(
+                port['network_id'])
+        return port_physnet_dict
+
+    def update_router_gateway_port_bindings(self, router, host):
+        status = (n_const.PORT_STATUS_ACTIVE if host
+                  else n_const.PORT_STATUS_DOWN)
+        context = n_context.get_admin_context()
+        filters = {'device_id': [router],
+                   'device_owner': [n_const.DEVICE_OWNER_ROUTER_GW]}
+        for port in self._plugin.get_ports(context, filters=filters):
+            # FIXME(lucasagomes): Ideally here we would use only
+            # one database transaction for the status and binding the
+            # host but, even tho update_port_status() receives a "host"
+            # parameter apparently it doesn't work for ports which the
+            # device owner is router_gateway. We need to look into it and
+            # fix the problem in Neutron before updating it here.
+            if host:
+                self._plugin.update_port(
+                    context, port['id'],
+                    {'port': {portbindings.HOST_ID: host}})
+
+            if port['status'] != status:
+                self._plugin.update_port_status(context, port['id'], status)
+
+    def schedule_unhosted_gateways(self):
+        port_physnet_dict = self._get_gateway_port_physnet_mapping()
+        chassis_physnets = self._sb_ovn.get_chassis_and_physnets()
+        cms = self._sb_ovn.get_gateway_chassis_from_cms_options()
+        unhosted_gateways = self._ovn.get_unhosted_gateways(
+            port_physnet_dict, chassis_physnets, cms)
+        for g_name in unhosted_gateways:
+            physnet = port_physnet_dict.get(g_name[len('lrp-'):])
+            # Remove any invalid gateway chassis from the list, otherwise
+            # we can have a situation where all existing_chassis are invalid
+            existing_chassis = self._ovn.get_gateway_chassis_binding(g_name)
+            master = existing_chassis[0] if existing_chassis else None
+            existing_chassis = self.scheduler.filter_existing_chassis(
+                nb_idl=self._ovn, gw_chassis=cms,
+                physnet=physnet, chassis_physnets=chassis_physnets,
+                existing_chassis=existing_chassis)
+            candidates = self._ovn_client.get_candidates_for_scheduling(
+                physnet, cms=cms, chassis_physnets=chassis_physnets)
+            chassis = self.scheduler.select(
+                self._ovn, self._sb_ovn, g_name, candidates=candidates,
+                existing_chassis=existing_chassis)
+            if master and master != chassis[0]:
+                if master not in chassis:
+                    LOG.debug("Master gateway chassis %(old)s "
+                              "has been removed from the system. Moving "
+                              "gateway %(gw)s to other chassis %(new)s.",
+                              {'gw': g_name,
+                               'old': master,
+                               'new': chassis[0]})
+                else:
+                    LOG.debug("Gateway %s is hosted at %s.", g_name, master)
+                    # NOTE(mjozefcz): It means scheduler moved master chassis
+                    # to other gw based on scheduling method. But we don't
+                    # want network flap - so moving actual master to be on
+                    # the top.
+                    index = chassis.index(master)
+                    chassis[0], chassis[index] = chassis[index], chassis[0]
+            # NOTE(dalvarez): Let's commit the changes in separate transactions
+            # as we will rely on those for scheduling subsequent gateways.
+            with self._ovn.transaction(check_error=True) as txn:
+                txn.add(self._ovn.update_lrouter_port(
+                    g_name, gateway_chassis=chassis))
+
+    @staticmethod
+    @registry.receives(resources.SUBNET, [events.AFTER_UPDATE])
+    def _subnet_update(resource, event, trigger, **kwargs):
+        l3plugin = directory.get_plugin(plugin_constants.L3)
+        if not l3plugin:
+            return
+        context = kwargs['context']
+        orig = kwargs['original_subnet']
+        current = kwargs['subnet']
+        orig_gw_ip = orig['gateway_ip']
+        current_gw_ip = current['gateway_ip']
+        if orig_gw_ip == current_gw_ip:
+            return
+        gw_ports = l3plugin._plugin.get_ports(context, filters={
+            'network_id': [orig['network_id']],
+            'device_owner': [n_const.DEVICE_OWNER_ROUTER_GW],
+            'fixed_ips': {'subnet_id': [orig['id']]},
+        })
+        router_ids = {port['device_id'] for port in gw_ports}
+        remove = [{'destination': '0.0.0.0/0', 'nexthop': orig_gw_ip}
+                  ] if orig_gw_ip else []
+        add = [{'destination': '0.0.0.0/0', 'nexthop': current_gw_ip}
+               ] if current_gw_ip else []
+        with l3plugin._ovn.transaction(check_error=True) as txn:
+            for router_id in router_ids:
+                l3plugin._ovn_client.update_router_routes(
+                    context, router_id, add, remove, txn=txn)
+
+    @staticmethod
+    @registry.receives(resources.PORT, [events.AFTER_UPDATE])
+    def _port_update(resource, event, trigger, **kwargs):
+        l3plugin = directory.get_plugin(plugin_constants.L3)
+        if not l3plugin:
+            return
+
+        current = kwargs['port']
+
+        if utils.is_lsp_router_port(current):
+            # We call the update_router port with if_exists, because neutron,
+            # internally creates the port, and then calls update, which will
+            # trigger this callback even before we had the chance to create
+            # the OVN NB DB side
+            l3plugin._ovn_client.update_router_port(current, if_exists=True)

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

@@ -0,0 +1,168 @@
+#    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 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__)
+
+OVN_QOS = 'qos'
+SUPPORTED_RULES = {
+    qos_consts.RULE_TYPE_BANDWIDTH_LIMIT: {
+        qos_consts.MAX_KBPS: {
+            'type:range': [0, db_consts.DB_INTEGER_MAX_VALUE]},
+        qos_consts.MAX_BURST: {
+            'type:range': [0, db_consts.DB_INTEGER_MAX_VALUE]},
+        qos_consts.DIRECTION: {
+            'type:values': [constants.EGRESS_DIRECTION]}
+    },
+}
+
+VIF_TYPES = [portbindings.VIF_TYPE_OVS, portbindings.VIF_TYPE_VHOST_USER]
+VNIC_TYPES = [portbindings.VNIC_NORMAL]
+
+
+class OVNQosNotificationDriver(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()
+
+    @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
+        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):
+                if rule.max_kbps:
+                    options['qos_max_rate'] = str(rule.max_kbps * 1000)
+                if rule.max_burst_kbps:
+                    options['qos_burst'] = str(rule.max_burst_kbps * 1000)
+        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/unit/scheduler/test_l3_ovn_scheduler.py

@@ -0,0 +1,228 @@
+#
+#    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 random
+
+import mock
+from neutron.tests import base
+
+from neutron.common.ovn import constants as ovn_const
+from neutron.scheduler import l3_ovn_scheduler
+
+
+class FakeOVNGatewaySchedulerNbOvnIdl(object):
+    def __init__(self, chassis_gateway_mapping, gateway):
+        self.get_all_chassis_gateway_bindings = mock.Mock(
+            return_value=chassis_gateway_mapping['Chassis_Bindings'])
+        self.get_gateway_chassis_binding = mock.Mock(
+            return_value=chassis_gateway_mapping['Gateways'].get(gateway,
+                                                                 None))
+
+
+class FakeOVNGatewaySchedulerSbOvnIdl(object):
+    def __init__(self, chassis_gateway_mapping):
+        self.get_all_chassis = mock.Mock(
+            return_value=chassis_gateway_mapping['Chassis'])
+
+
+class TestOVNGatewayScheduler(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestOVNGatewayScheduler, self).setUp()
+
+        # Overwritten by derived classes
+        self.l3_scheduler = None
+
+        # Used for unit tests
+        self.new_gateway_name = 'lrp_new'
+        self.fake_chassis_gateway_mappings = {
+            'None': {'Chassis': [],
+                     'Gateways': {
+                         'g1': [ovn_const.OVN_GATEWAY_INVALID_CHASSIS]}},
+            'Multiple1': {'Chassis': ['hv1', 'hv2', 'hv3', 'hv4', 'hv5'],
+                          'Gateways': {'g1': ['hv1', 'hv2', 'hv3', 'hv4'],
+                                       'g2': ['hv1', 'hv2', 'hv3'],
+                                       'g3': ['hv1', 'hv2'],
+                                       'g4': ['hv1']}},
+            'Multiple2': {'Chassis': ['hv1', 'hv2', 'hv3'],
+                          'Gateways': {'g1': ['hv1'],
+                                       'g2': ['hv1'],
+                                       'g3': ['hv1']}},
+            'Multiple3': {'Chassis': ['hv1', 'hv2', 'hv3'],
+                          'Gateways': {'g1': ['hv3'],
+                                       'g2': ['hv2'],
+                                       'g3': ['hv2']}},
+            'Multiple4': {'Chassis': ['hv1', 'hv2'],
+                          'Gateways': {'g1': ['hv1'],
+                                       'g2': ['hv1'],
+                                       'g3': ['hv1'],
+                                       'g4': ['hv1'],
+                                       'g5': ['hv1'],
+                                       'g6': ['hv1']}}}
+
+        # Determine the chassis to gateway list bindings
+        for details in self.fake_chassis_gateway_mappings.values():
+            self.assertNotIn(self.new_gateway_name, details['Gateways'])
+            details.setdefault('Chassis_Bindings', {})
+            for chassis in details['Chassis']:
+                details['Chassis_Bindings'].setdefault(chassis, [])
+            for gw, chassis_list in details['Gateways'].items():
+                for chassis in chassis_list:
+                    if chassis in details['Chassis_Bindings']:
+                        details['Chassis_Bindings'][chassis].append((gw, 0))
+
+    def select(self, chassis_gateway_mapping, gateway_name):
+        nb_idl = FakeOVNGatewaySchedulerNbOvnIdl(chassis_gateway_mapping,
+                                                 gateway_name)
+        sb_idl = FakeOVNGatewaySchedulerSbOvnIdl(chassis_gateway_mapping)
+        return self.l3_scheduler.select(nb_idl, sb_idl,
+                                        gateway_name)
+
+    def filter_existing_chassis(self, *args, **kwargs):
+        return self.l3_scheduler.filter_existing_chassis(
+            nb_idl=kwargs.pop('nb_idl'), gw_chassis=kwargs.pop('gw_chassis'),
+            physnet=kwargs.pop('physnet'),
+            chassis_physnets=kwargs.pop('chassis_physnets'),
+            existing_chassis=kwargs.pop('existing_chassis'))
+
+
+class OVNGatewayChanceScheduler(TestOVNGatewayScheduler):
+
+    def setUp(self):
+        super(OVNGatewayChanceScheduler, self).setUp()
+        self.l3_scheduler = l3_ovn_scheduler.OVNGatewayChanceScheduler()
+
+    def test_no_chassis_available_for_existing_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['None']
+        gateway_name = random.choice(list(mapping['Gateways'].keys()))
+        chassis = self.select(mapping, gateway_name)
+        self.assertEqual([ovn_const.OVN_GATEWAY_INVALID_CHASSIS], chassis)
+
+    def test_no_chassis_available_for_new_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['None']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        self.assertEqual([ovn_const.OVN_GATEWAY_INVALID_CHASSIS], chassis)
+
+    def test_random_chassis_available_for_new_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple1']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        self.assertItemsEqual(chassis, mapping.get('Chassis'))
+
+    def test_filter_existing_chassis(self):
+        # filter_existing_chassis is scheduler independent, but calling
+        # it from Base class didnt seem right. Also, there is no need to have
+        # another test in LeastLoadedScheduler.
+        chassis_physnets = {'temp': ['phys-network-0', 'phys-network-1']}
+        nb_idl = FakeOVNGatewaySchedulerNbOvnIdl(
+            self.fake_chassis_gateway_mappings['None'], 'g1')
+        # Check if invalid chassis is removed
+        self.assertEqual(
+            ['temp'], self.filter_existing_chassis(
+                nb_idl=nb_idl, gw_chassis=["temp"],
+                physnet='phys-network-1',
+                chassis_physnets=chassis_physnets,
+                existing_chassis=['temp',
+                                  ovn_const.OVN_GATEWAY_INVALID_CHASSIS]))
+        # Check if invalid is removed -II
+        self.assertFalse(
+            self.filter_existing_chassis(
+                nb_idl=nb_idl, gw_chassis=["temp"],
+                physnet='phys-network-1',
+                chassis_physnets=chassis_physnets,
+                existing_chassis=[ovn_const.OVN_GATEWAY_INVALID_CHASSIS]))
+        # Check if chassis removed when physnet doesnt exist
+        self.assertFalse(
+            self.filter_existing_chassis(
+                nb_idl=nb_idl, gw_chassis=["temp"],
+                physnet='phys-network-2',
+                chassis_physnets=chassis_physnets,
+                existing_chassis=['temp']))
+        # Check if chassis removed when it doesnt exist in gw_chassis
+        # or in chassis_physnets
+        self.assertFalse(
+            self.filter_existing_chassis(
+                nb_idl=nb_idl, gw_chassis=["temp1"],
+                physnet='phys-network-2',
+                chassis_physnets=chassis_physnets,
+                existing_chassis=['temp']))
+
+
+class OVNGatewayLeastLoadedScheduler(TestOVNGatewayScheduler):
+
+    def setUp(self):
+        super(OVNGatewayLeastLoadedScheduler, self).setUp()
+        self.l3_scheduler = l3_ovn_scheduler.OVNGatewayLeastLoadedScheduler()
+
+    def test_no_chassis_available_for_existing_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['None']
+        gateway_name = random.choice(list(mapping['Gateways'].keys()))
+        chassis = self.select(mapping, gateway_name)
+        self.assertEqual([ovn_const.OVN_GATEWAY_INVALID_CHASSIS], chassis)
+
+    def test_no_chassis_available_for_new_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['None']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        self.assertEqual([ovn_const.OVN_GATEWAY_INVALID_CHASSIS], chassis)
+
+    def test_least_loaded_chassis_available_for_new_gateway1(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple1']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        self.assertItemsEqual(chassis, mapping.get('Chassis'))
+        # least loaded will be the first one in the list,
+        # networking-ovn will assign highest priority to this first element
+        self.assertEqual(['hv5', 'hv4', 'hv3', 'hv2', 'hv1'], chassis)
+
+    def test_least_loaded_chassis_available_for_new_gateway2(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple2']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        # hv1 will have least priority
+        self.assertEqual(chassis[2], 'hv1')
+
+    def test_least_loaded_chassis_available_for_new_gateway3(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple3']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        # least loaded chassis will be in the front of the list
+        self.assertEqual(['hv1', 'hv3', 'hv2'], chassis)
+
+    def test_least_loaded_chassis_with_rebalance(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple4']
+        gateway_name = self.new_gateway_name
+        chassis = self.select(mapping, gateway_name)
+        # least loaded chassis will be in the front of the list
+        self.assertEqual(['hv2', 'hv1'], chassis)
+
+    def test_existing_chassis_available_for_existing_gateway(self):
+        mapping = self.fake_chassis_gateway_mappings['Multiple1']
+        gateway_name = random.choice(list(mapping['Gateways'].keys()))
+        chassis = self.select(mapping, gateway_name)
+        self.assertEqual(ovn_const.MAX_GW_CHASSIS, len(chassis))
+
+    def test__get_chassis_load_by_prios_several_ports(self):
+        # Adding 5 ports of prio 1 and 5 ports of prio 2
+        chassis_info = []
+        for i in range(1, 6):
+            chassis_info.append(('lrp', 1))
+            chassis_info.append(('lrp', 2))
+        actual = self.l3_scheduler._get_chassis_load_by_prios(chassis_info)
+        expected = {1: 5, 2: 5}
+        self.assertItemsEqual(expected.items(), actual)
+
+    def test__get_chassis_load_by_prios_no_ports(self):
+        self.assertFalse(self.l3_scheduler._get_chassis_load_by_prios([]))

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

@@ -0,0 +1,242 @@
+#    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.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):
+
+    def setUp(self):
+        super(TestOVNQosNotificationDriver, 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.policy = "policy"
+
+    def test_create_policy(self):
+        self.driver.create_policy(context, self.policy)
+        self.driver._driver._ovn_client._qos_driver.create_policy.\
+            assert_not_called()
+
+    def test_update_policy(self):
+        self.driver.update_policy(context, self.policy)
+        self.driver._driver._ovn_client._qos_driver.update_policy.\
+            assert_called_once_with(context, self.policy)
+
+    def test_delete_policy(self):
+        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.rule = self._create_bw_limit_rule()
+        self.expected = {'qos_max_rate': '1000', 'qos_burst': '100000'}
+
+    def _create_bw_limit_rule(self):
+        rule_obj = qos_rule.QosBandwidthLimitRule()
+        rule_obj.id = uuidutils.generate_uuid()
+        rule_obj.max_kbps = 1
+        rule_obj.max_burst_kbps = 100
+        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_rule(self):
+        self._generate_port_options(self.policy_id, [self.rule], self.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={})

setup.cfg

@@ -78,6 +78,7 @@ neutron.service_plugins =
     port_forwarding = neutron.services.portforwarding.pf_plugin:PortForwardingPlugin
     placement = neutron.services.placement_report.plugin:PlacementReportPlugin
     conntrack_helper = neutron.services.conntrack_helper.plugin:Plugin
+    ovn-router = neutron.services.ovn_l3.plugin:OVNL3RouterPlugin
 neutron.ml2.type_drivers =
     flat = neutron.plugins.ml2.drivers.type_flat:FlatTypeDriver
     local = neutron.plugins.ml2.drivers.type_local:LocalTypeDriver