Merge "Transition qos notification driver into qos driver"

doc/source/devref/quality_of_service.rst

@@ -45,18 +45,15 @@ Service side design
   QoSPlugin, service plugin that implements 'qos' extension, receiving and
   handling API calls to create/modify policies and rules.
 
-* neutron.services.qos.notification_drivers.manager:
+* neutron.services.qos.drivers.manager:
-  the manager that passes object notifications down to every enabled
+  the manager that passes object actions down to every enabled QoS driver and
-  notification driver.
+  issues RPC calls when any of the drivers require RPC push notifications.
 
-* neutron.services.qos.notification_drivers.qos_base:
+* neutron.services.qos.drivers.base:
-  the interface class for pluggable notification drivers that are used to
+  the interface class for pluggable QoS drivers that are used to update
-  update backends about new {create, update, delete} events on any rule or
+  backends about new {create, update, delete} events on any rule or policy
-  policy change.
+  change. The drivers also declare which QoS rules, VIF drivers and VNIC
-
+  types are supported.
-* neutron.services.qos.notification_drivers.message_queue:
-  MQ-based reference notification driver which updates agents via messaging
-  bus, using `RPC callbacks <rpc_callbacks.html>`_.
 
 * neutron.core_extensions.base:
   Contains an interface class to implement core resource (port/network)
@@ -96,21 +93,19 @@ NotImplemented.
 Supported QoS rule types
 ~~~~~~~~~~~~~~~~~~~~~~~~
 
-Any plugin or Ml2 mechanism driver can claim support for some QoS rule types by
+Each QoS driver has a property called supported_rule_types, where the driver
-providing a plugin/driver class property called 'supported_qos_rule_types' that
+exposes the rules it's able to handle.
-should return a list of strings that correspond to QoS rule types (for the list
-of all rule types, see: neutron.services.qos.qos_consts.VALID_RULE_TYPES).
 
-In the most simple case, the property can be represented by a simple Python
+For a list of all rule types, see:
-list defined on the class.
+neutron.services.qos.qos_consts.VALID_RULE_TYPES.
 
-For Ml2 plugin, the list of supported QoS rule types is defined as a common
+The list of supported QoS rule types exposed by neutron is calculated as
-subset of rules supported by all active mechanism drivers.
+the common subset of rules supported by all active QoS drivers.
 
 Note: the list of supported rule types reported by core plugin is not enforced
 when accessing QoS rule resources. This is mostly because then we would not be
-able to create any rules while at least one ml2 driver in gate lacks support
+able to create rules while at least one of the QoS driver in gate lacks
-for QoS (at the moment of writing, linuxbridge is such a driver).
+support for the rules we're trying to test.
 
 
 Database models
@@ -401,15 +396,15 @@ Traffic in the tap port is redirected (mirrored) to the IFB using a Traffic
 Control filter
 (`Filter Actions <http://linux-ip.net/gl/tc-filters/tc-filters-node2.html>`_).
 
-Notification driver design
+QoS driver design
---------------------------
+-----------------
 
 QoS framework is flexible enough to support any third-party vendor. To integrate a
 third party driver (that just wants to be aware of the QoS create/update/delete API
-calls), one needs to implement 'neutron.services.qos.notification_drivers.qos_base',
+calls), one needs to implement 'neutron.services.qos.drivers.base', and register
-register its specific driver information to the 'notification_drivers' stevedore
+the driver during the core plugin or mechanism driver load, see
-namespace in the setup.cfg and finally set the 'notification_drivers' parameter in
+
-the [qos] section of the neutron config file.
+neutron.services.qos.drivers.openvswitch.driver register method for an example.
 
 .. note::
  All the functionality MUST be implemented by the vendor, neutron's QoS framework
@@ -424,7 +419,6 @@ To enable the service, the following steps should be followed:
 On server side:
 
 * enable qos service in service_plugins;
-* set the needed notification_drivers in [qos] section (message_queue is the default);
 * for ml2, add 'qos' to extension_drivers in [ml2] section.
 
 On agent side (OVS):

neutron/conf/services/qos_driver_manager.py

@@ -16,7 +16,9 @@ from neutron._i18n import _
 QOS_PLUGIN_OPTS = [
     cfg.ListOpt('notification_drivers',
                 default=['message_queue'],
-                help=_('Drivers list to use to send the update notification')),
+                help=_("Drivers list to use to send the update notification. "
+                       "This option will be unused in Pike."),
+                deprecated_for_removal=True),
 ]
 
 

neutron/objects/qos/rule_type.py

@@ -10,13 +10,18 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from neutron.plugins.common import constants
 from neutron_lib.plugins import directory
+from oslo_log import log as logging
 from oslo_versionedobjects import base as obj_base
 from oslo_versionedobjects import fields as obj_fields
 
+from neutron._i18n import _LW
 from neutron.objects import base
 from neutron.services.qos import qos_consts
 
+LOG = logging.getLogger(__name__)
+
 
 class RuleTypeField(obj_fields.BaseEnumField):
 
@@ -42,7 +47,22 @@ class QosRuleType(base.NeutronObject):
     def get_objects(cls, validate_filters=True, **kwargs):
         if validate_filters:
             cls.validate_filters(**kwargs)
-        core_plugin = directory.get_plugin()
+
+        #TODO(mangelajo): remove in backwards compatible available rule
+        #                 inspection in Pike
+
+        core_plugin_supported_rules = getattr(
+            directory.get_plugin(), 'supported_qos_rule_types', None)
+
+        rule_types = (
+            core_plugin_supported_rules or
+            directory.get_plugin(alias=constants.QOS).supported_rule_types)
+
+        if core_plugin_supported_rules:
+            LOG.warning(_LW(
+                "Your core plugin defines supported_qos_rule_types which is "
+                "deprecated and shall be implemented through a QoS driver."
+            ))
+
         # TODO(ihrachys): apply filters to returned result
-        return [cls(type=type_)
+        return [cls(type=type_) for type_ in rule_types]
-                for type_ in core_plugin.supported_qos_rule_types]

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

@@ -22,8 +22,7 @@ from neutron.agent.l2.extensions import qos_linux as qos
 from neutron.agent.linux import iptables_manager
 from neutron.agent.linux import tc_lib
 import neutron.common.constants as const
-from neutron.plugins.ml2.drivers.linuxbridge.mech_driver import (
+from neutron.services.qos.drivers.linuxbridge import driver
-    mech_linuxbridge)
 from neutron.services.qos import qos_consts
 
 LOG = log.getLogger(__name__)
@@ -36,9 +35,7 @@ class QosLinuxbridgeAgentDriver(qos.QosLinuxAgentDriver):
     #     the delete function, to have the 'direction' parameter. This QoS
     #     extension modification is going to be implemented in
     #     https://review.openstack.org/#/c/341186/
-    SUPPORTED_RULES = (
+    SUPPORTED_RULES = driver.SUPPORTED_RULES
-        mech_linuxbridge.LinuxbridgeMechanismDriver.supported_qos_rule_types
-    )
 
     IPTABLES_DIRECTION = {const.INGRESS_DIRECTION: 'physdev-out',
                           const.EGRESS_DIRECTION: 'physdev-in'}

neutron/plugins/ml2/drivers/linuxbridge/mech_driver/mech_linuxbridge.py

@@ -19,7 +19,7 @@ from neutron.agent import securitygroups_rpc
 from neutron.extensions import portbindings
 from neutron.plugins.common import constants as p_constants
 from neutron.plugins.ml2.drivers import mech_agent
-from neutron.services.qos import qos_consts
+from neutron.services.qos.drivers.linuxbridge import driver as lb_qos_driver
 
 
 class LinuxbridgeMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
@@ -32,16 +32,13 @@ class LinuxbridgeMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
     network.
     """
 
-    supported_qos_rule_types = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
-                                qos_consts.RULE_TYPE_DSCP_MARKING,
-                                qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]
-
     def __init__(self):
         sg_enabled = securitygroups_rpc.is_firewall_enabled()
         super(LinuxbridgeMechanismDriver, self).__init__(
             constants.AGENT_TYPE_LINUXBRIDGE,
             portbindings.VIF_TYPE_BRIDGE,
             {portbindings.CAP_PORT_FILTER: sg_enabled})
+        lb_qos_driver.register()
 
     def get_allowed_network_types(self, agent):
         return (agent['configurations'].get('tunnel_types', []) +

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

@@ -19,16 +19,14 @@ from neutron.agent.l2.extensions import qos_linux as qos
 from neutron.plugins.ml2.drivers.mech_sriov.agent.common import (
     exceptions as exc)
 from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
-from neutron.plugins.ml2.drivers.mech_sriov.mech_driver import (
+from neutron.services.qos.drivers.sriov import driver
-    mech_driver)
 
 LOG = logging.getLogger(__name__)
 
 
 class QosSRIOVAgentDriver(qos.QosLinuxAgentDriver):
 
-    SUPPORTED_RULES = (
+    SUPPORTED_RULES = driver.SUPPORTED_RULES
-        mech_driver.SriovNicSwitchMechanismDriver.supported_qos_rule_types)
 
     def __init__(self):
         super(QosSRIOVAgentDriver, self).__init__()

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

@@ -23,7 +23,7 @@ from neutron.plugins.ml2 import driver_api as api
 from neutron.plugins.ml2.drivers import mech_agent
 from neutron.plugins.ml2.drivers.mech_sriov.mech_driver \
     import exceptions as exc
-from neutron.services.qos import qos_consts
+from neutron.services.qos.drivers.sriov import driver as sriov_qos_driver
 
 
 LOG = log.getLogger(__name__)
@@ -43,11 +43,6 @@ class SriovNicSwitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
     L2 Agent presents in  order to manage port update events.
     """
 
-    supported_qos_rule_types = [
-        qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
-        qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH,
-    ]
-
     def __init__(self,
                  agent_type=constants.AGENT_TYPE_NIC_SWITCH,
                  vif_details={portbindings.CAP_PORT_FILTER: False},
@@ -69,6 +64,7 @@ class SriovNicSwitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
                 else portbindings.VIF_TYPE_HW_VEB
              for vtype in self.supported_vnic_types})
         self.vif_details = vif_details
+        sriov_qos_driver.register()
 
     def get_allowed_network_types(self, agent):
         return (p_const.TYPE_FLAT, p_const.TYPE_VLAN)

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

@@ -18,8 +18,7 @@ from oslo_config import cfg
 from oslo_log import log as logging
 
 from neutron.agent.l2.extensions import qos_linux as qos
-from neutron.plugins.ml2.drivers.openvswitch.mech_driver import (
+from neutron.services.qos.drivers.openvswitch import driver
-    mech_openvswitch)
 from neutron.services.qos import qos_consts
 
 
@@ -28,8 +27,7 @@ LOG = logging.getLogger(__name__)
 
 class QosOVSAgentDriver(qos.QosLinuxAgentDriver):
 
-    SUPPORTED_RULES = (
+    SUPPORTED_RULES = driver.SUPPORTED_RULES
-        mech_openvswitch.OpenvswitchMechanismDriver.supported_qos_rule_types)
 
     def __init__(self):
         super(QosOVSAgentDriver, self).__init__()

neutron/plugins/ml2/drivers/openvswitch/mech_driver/mech_openvswitch.py

@@ -27,7 +27,8 @@ from neutron.plugins.ml2 import driver_api as api
 from neutron.plugins.ml2.drivers import mech_agent
 from neutron.plugins.ml2.drivers.openvswitch.agent.common \
     import constants as a_const
-from neutron.services.qos import qos_consts
+from neutron.services.qos.drivers.openvswitch import driver as ovs_qos_driver
+
 
 IPTABLES_FW_DRIVER_FULL = ("neutron.agent.linux.iptables_firewall."
                            "OVSHybridIptablesFirewallDriver")
@@ -43,9 +44,6 @@ class OpenvswitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
     network.
     """
 
-    supported_qos_rule_types = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
-                                qos_consts.RULE_TYPE_DSCP_MARKING]
-
     def __init__(self):
         sg_enabled = securitygroups_rpc.is_firewall_enabled()
         hybrid_plug_required = (not cfg.CONF.SECURITYGROUP.firewall_driver or
@@ -57,6 +55,7 @@ class OpenvswitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
             constants.AGENT_TYPE_OVS,
             portbindings.VIF_TYPE_OVS,
             vif_details)
+        ovs_qos_driver.register()
 
     def get_allowed_network_types(self, agent):
         return (agent['configurations'].get('tunnel_types', []) +

neutron/plugins/ml2/managers.py

@@ -33,7 +33,6 @@ from neutron.extensions import vlantransparent
 from neutron.plugins.ml2.common import exceptions as ml2_exc
 from neutron.plugins.ml2 import driver_api as api
 from neutron.plugins.ml2 import models
-from neutron.services.qos import qos_consts
 
 LOG = log.getLogger(__name__)
 
@@ -369,48 +368,6 @@ class MechanismManager(stevedore.named.NamedExtensionManager):
         LOG.info(_LI("Registered mechanism drivers: %s"),
                  [driver.name for driver in self.ordered_mech_drivers])
 
-    @property
-    def supported_qos_rule_types(self):
-        if not self.ordered_mech_drivers:
-            return []
-
-        rule_types = set(qos_consts.VALID_RULE_TYPES)
-        binding_driver_found = False
-
-        # Recalculate on every call to allow drivers determine supported rule
-        # types dynamically
-        for driver in self.ordered_mech_drivers:
-            driver_obj = driver.obj
-            if driver_obj._supports_port_binding:
-                binding_driver_found = True
-                if hasattr(driver_obj, 'supported_qos_rule_types'):
-                    new_rule_types = \
-                        rule_types & set(driver_obj.supported_qos_rule_types)
-                    dropped_rule_types = new_rule_types - rule_types
-                    if dropped_rule_types:
-                        LOG.info(
-                            _LI("%(rule_types)s rule types disabled for ml2 "
-                                "because %(driver)s does not support them"),
-                            {'rule_types': ', '.join(dropped_rule_types),
-                             'driver': driver.name})
-                    rule_types = new_rule_types
-                else:
-                    # at least one of drivers does not support QoS, meaning
-                    # there are no rule types supported by all of them
-                    LOG.warning(
-                        _LW("%s does not support QoS; "
-                            "no rule types available"),
-                        driver.name)
-                    return []
-
-        if binding_driver_found:
-            rule_types = list(rule_types)
-        else:
-            rule_types = []
-        LOG.debug("Supported QoS rule types "
-                  "(common subset for all mech drivers): %s", rule_types)
-        return rule_types
-
     def initialize(self):
         for driver in self.ordered_mech_drivers:
             LOG.info(_LI("Initializing mechanism driver '%s'"), driver.name)

neutron/plugins/ml2/plugin.py

@@ -236,10 +236,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
                 return
         self.update_port_status(context, port_id, const.PORT_STATUS_ACTIVE)
 
-    @property
-    def supported_qos_rule_types(self):
-        return self.mechanism_manager.supported_qos_rule_types
-
     @log_helpers.log_method_call
     def _start_rpc_notifiers(self):
         """Initialize RPC notifiers for agents."""

neutron/services/qos/drivers/base.py

@@ -0,0 +1,100 @@
+# Copyright 2016 Hewlett Packard Enterprise Development Company, LP
+# Copyright 2016 Red Hat Inc.
+#
+# All Rights Reserved.
+#
+# 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.callbacks import events
+from neutron.callbacks import registry
+from neutron.services.qos import qos_consts
+
+
+class DriverBase(object):
+
+    def __init__(self, name, vif_types, vnic_types,
+                 supported_rules,
+                 requires_rpc_notifications=False):
+        """Instantiate a qos driver.
+
+        :param name: driver name.
+        :param vif_types: list of interfaces (VIFs) supported.
+        :param vnic_types: list of vnic types supported.
+        :param supported_rules: list of supported rules.
+        :param requires_rpc_notifications: indicates if this driver
+               expects rpc push notifications to be sent from the driver.
+        """
+
+        self.name = name
+        self.vif_types = vif_types
+        self.vnic_types = vnic_types
+        self.supported_rules = supported_rules
+        self.requires_rpc_notifications = requires_rpc_notifications
+        registry.subscribe(self._register,
+                           qos_consts.QOS_PLUGIN,
+                           events.AFTER_INIT)
+
+    def _register(self, resource, event, trigger, **kwargs):
+        if self.is_loaded:
+            # trigger is the QosServiceDriverManager
+            trigger.register_driver(self)
+
+    def is_loaded(self):
+        """True if the driver is active for the Neutron Server.
+
+        Implement this property to determine if your driver is actively
+        configured for this Neutron Server deployment.
+        """
+        return True
+
+    def is_vif_type_compatible(self, vif_type):
+        """True if the driver is compatible with the VIF type."""
+        return vif_type in self.vif_types
+
+    def is_vnic_compatible(self, vnic_type):
+        """True if the driver is compatible with the specific VNIC type."""
+        return vnic_type in self.vnic_types
+
+    def is_rule_supported(self, rule):
+        return rule.rule_type in self.supported_rules
+
+    def create_policy(self, context, policy):
+        """Create policy invocation.
+
+        This method can be implemented by the specific driver subclass
+        to update the backend where necessary with the specific policy
+        information.
+
+        :param context: current running context information
+        :param policy: a QoSPolicy object being created, which will have no
+                      rules.
+        """
+
+    def update_policy(self, context, policy):
+        """Update policy invocation.
+
+        This method can be implemented by the specific driver subclass
+        to update the backend where necessary.
+
+        :param context: current running context information
+        :param policy: a QoSPolicy object being updated.
+        """
+
+    def delete_policy(self, context, policy):
+        """Delete policy invocation.
+
+        This method can be implemented by the specific driver subclass
+        to delete the backend policy where necessary.
+
+        :param context: current running context information
+        :param policy: a QoSPolicy object being deleted
+        """

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

@@ -0,0 +1,48 @@
+# Copyright (c) 2016 Red Hat Inc.
+# All Rights Reserved.
+#
+#    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 oslo_log import log as logging
+
+from neutron.extensions import portbindings
+from neutron.services.qos.drivers import base
+from neutron.services.qos import qos_consts
+
+LOG = logging.getLogger(__name__)
+
+DRIVER = None
+
+SUPPORTED_RULES = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
+                   qos_consts.RULE_TYPE_DSCP_MARKING,
+                   qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]
+
+
+class LinuxBridgeDriver(base.DriverBase):
+
+    @staticmethod
+    def create():
+        return LinuxBridgeDriver(
+            name='linuxbridge',
+            vif_types=[portbindings.VIF_TYPE_BRIDGE],
+            vnic_types=[portbindings.VNIC_NORMAL],
+            supported_rules=SUPPORTED_RULES,
+            requires_rpc_notifications=True)
+
+
+def register():
+    """Register the driver."""
+    global DRIVER
+    if not DRIVER:
+        DRIVER = LinuxBridgeDriver.create()
+    LOG.debug('Linuxbridge QoS driver registered')

neutron/services/qos/drivers/manager.py

@@ -0,0 +1,117 @@
+#    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 oslo_log import log as logging
+from oslo_utils import excutils
+
+
+from neutron._i18n import _LE, _LW
+from neutron.api.rpc.callbacks import events as rpc_events
+from neutron.api.rpc.callbacks.producer import registry as rpc_registry
+from neutron.api.rpc.callbacks import resources
+from neutron.api.rpc.handlers import resources_rpc
+from neutron.callbacks import events
+from neutron.callbacks import registry
+from neutron.conf.services import qos_driver_manager as qos_mgr
+from neutron.objects.qos import policy as policy_object
+from neutron.services.qos import qos_consts
+
+qos_mgr.register_qos_plugin_opts()
+
+LOG = logging.getLogger(__name__)
+
+
+class QosServiceDriverManager(object):
+
+    def __init__(self, enable_rpc=False):
+        self._drivers = []
+        self.notification_api = resources_rpc.ResourcesPushRpcApi()
+        #TODO(mangelajo): remove the enable_rpc parameter in Pike since
+        #                 we only use it when a message_queue derived driver
+        #                 is found in the notification_drivers
+        self.rpc_notifications_required = enable_rpc
+        rpc_registry.provide(self._get_qos_policy_cb, resources.QOS_POLICY)
+        # notify any registered QoS driver that we're ready, those will
+        # call the driver manager back with register_driver if they
+        # are enabled
+        registry.notify(qos_consts.QOS_PLUGIN, events.AFTER_INIT, self)
+
+        if self.rpc_notifications_required:
+            self.push_api = resources_rpc.ResourcesPushRpcApi()
+
+    @staticmethod
+    def _get_qos_policy_cb(resource, policy_id, **kwargs):
+        context = kwargs.get('context')
+        if context is None:
+            LOG.warning(_LW(
+                'Received %(resource)s %(policy_id)s without context'),
+                {'resource': resource, 'policy_id': policy_id}
+            )
+            return
+
+        policy = policy_object.QosPolicy.get_object(context, id=policy_id)
+        return policy
+
+    def call(self, method_name, *args, **kwargs):
+        """Helper method for calling a method across all extension drivers."""
+        for driver in self._drivers:
+            try:
+                getattr(driver, method_name)(*args, **kwargs)
+            except Exception:
+                with excutils.save_and_reraise_exception():
+                    LOG.exception(_LE("Extension driver '%(name)s' failed in "
+                                      "%(method)s"),
+                                  {'name': driver.name, 'method': method_name})
+
+        if self.rpc_notifications_required:
+            context = kwargs.get('context') or args[0]
+            policy_obj = kwargs.get('policy_obj') or args[1]
+
+            # we don't push create_policy events since policies are empty
+            # on creation, they only become of any use when rules get
+            # attached to them.
+            if method_name == 'update_policy':
+                self.push_api.push(context, [policy_obj], rpc_events.UPDATED)
+
+            elif method_name == 'delete_policy':
+                self.push_api.push(context, [policy_obj], rpc_events.DELETED)
+
+    def register_driver(self, driver):
+        """Register driver with qos plugin.
+
+        This method is called from drivers on INIT event.
+        """
+        self._drivers.append(driver)
+        self.rpc_notifications_required |= driver.requires_rpc_notifications
+
+    @property
+    def supported_rule_types(self):
+        if not self._drivers:
+            return []
+
+        rule_types = set(qos_consts.VALID_RULE_TYPES)
+
+        # Recalculate on every call to allow drivers determine supported rule
+        # types dynamically
+        for driver in self._drivers:
+            new_rule_types = rule_types & set(driver.supported_rules)
+            dropped_rule_types = rule_types - new_rule_types
+            if dropped_rule_types:
+                LOG.debug("%(rule_types)s rule types disabled "
+                          "because enabled %(driver)s does not support them",
+                          {'rule_types': ', '.join(dropped_rule_types),
+                           'driver': driver.name})
+            rule_types = new_rule_types
+
+        LOG.debug("Supported QoS rule types "
+                  "(common subset for all loaded QoS drivers): %s", rule_types)
+        return rule_types

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

@@ -0,0 +1,48 @@
+# Copyright (c) 2016 Red Hat Inc.
+# All Rights Reserved.
+#
+#    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 oslo_log import log as logging
+
+from neutron.extensions import portbindings
+from neutron.services.qos.drivers import base
+from neutron.services.qos import qos_consts
+
+LOG = logging.getLogger(__name__)
+
+DRIVER = None
+
+SUPPORTED_RULES = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
+                   qos_consts.RULE_TYPE_DSCP_MARKING]
+
+
+class OVSDriver(base.DriverBase):
+
+    @staticmethod
+    def create():
+        return OVSDriver(
+            name='openvswitch',
+            vif_types=[portbindings.VIF_TYPE_OVS,
+                       portbindings.VIF_TYPE_VHOST_USER],
+            vnic_types=[portbindings.VNIC_NORMAL],
+            supported_rules=SUPPORTED_RULES,
+            requires_rpc_notifications=True)
+
+
+def register():
+    """Register the driver."""
+    global DRIVER
+    if not DRIVER:
+        DRIVER = OVSDriver.create()
+    LOG.debug('Open vSwitch QoS driver registered')

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

@@ -0,0 +1,48 @@
+# Copyright (c) 2016 Red Hat Inc.
+# All Rights Reserved.
+#
+#    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 oslo_log import log as logging
+
+from neutron.extensions import portbindings
+from neutron.services.qos.drivers import base
+from neutron.services.qos import qos_consts
+
+LOG = logging.getLogger(__name__)
+
+DRIVER = None
+
+SUPPORTED_RULES = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
+                   qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]
+
+
+class SRIOVNICSwitchDriver(base.DriverBase):
+
+    @staticmethod
+    def create():
+        return SRIOVNICSwitchDriver(
+            name='sriovnicswitch',
+            vif_types=[portbindings.VIF_TYPE_HW_VEB],
+            vnic_types=[portbindings.VNIC_DIRECT,
+                        portbindings.VNIC_MACVTAP],
+            supported_rules=SUPPORTED_RULES,
+            requires_rpc_notifications=True)
+
+
+def register():
+    """Register the driver."""
+    global DRIVER
+    if not DRIVER:
+        DRIVER = SRIOVNICSwitchDriver.create()
+    LOG.debug('SR-IOV NIC Switch QoS driver registered')

neutron/services/qos/notification_drivers/manager.py

@@ -12,9 +12,10 @@
 from oslo_config import cfg
 from oslo_log import log as logging
 
+from neutron._i18n import _LI
 from neutron.conf.services import qos_driver_manager as qos_mgr
-from neutron._i18n import _, _LI
 from neutron import manager
+from neutron.services.qos.notification_drivers import message_queue
 
 QOS_DRIVER_NAMESPACE = 'neutron.qos.notification_drivers'
 qos_mgr.register_qos_plugin_opts()
@@ -40,14 +41,23 @@ class QosServiceNotificationDriverManager(object):
         for driver in self.notification_drivers:
             driver.create_policy(context, qos_policy)
 
+    @property
+    def has_message_queue_driver(self):
+        """Determine if we have any message_queue derived driver in the
+        notifications drivers, so the QoS plugin will forcefully enable
+        the rpc notifications for Pike, since our message_queue driver
+        is a dummy which doesn't send any messages.
+        """
+        #TODO(mangelajo): remove this in Pike
+        return any(
+            isinstance(driver, message_queue.RpcQosServiceNotificationDriver)
+            for driver in self.notification_drivers)
+
     def _load_drivers(self, notification_drivers):
         """Load all the instances of the configured QoS notification drivers
 
         :param notification_drivers: comma separated string
         """
-        if not notification_drivers:
-            raise SystemExit(_('A QoS driver must be specified'))
-        LOG.debug("Loading QoS notification drivers: %s", notification_drivers)
         for notification_driver in notification_drivers:
             driver_ins = self._load_driver_instance(notification_driver)
             self.notification_drivers.append(driver_ins)

neutron/services/qos/notification_drivers/message_queue.py

@@ -13,47 +13,29 @@
 from oslo_log import log as logging
 
 from neutron._i18n import _LW
-from neutron.api.rpc.callbacks import events
-from neutron.api.rpc.callbacks.producer import registry
-from neutron.api.rpc.callbacks import resources
-from neutron.api.rpc.handlers import resources_rpc
-from neutron.objects.qos import policy as policy_object
 from neutron.services.qos.notification_drivers import qos_base
 
 
 LOG = logging.getLogger(__name__)
 
 
-def _get_qos_policy_cb(resource, policy_id, **kwargs):
-    context = kwargs.get('context')
-    if context is None:
-        LOG.warning(_LW(
-            'Received %(resource)s %(policy_id)s without context'),
-            {'resource': resource, 'policy_id': policy_id}
-        )
-        return
-
-    policy = policy_object.QosPolicy.get_object(context, id=policy_id)
-    return policy
-
-
 class RpcQosServiceNotificationDriver(
     qos_base.QosServiceNotificationDriverBase):
     """RPC message queue service notification driver for QoS."""
 
     def __init__(self):
-        self.notification_api = resources_rpc.ResourcesPushRpcApi()
+        LOG.warning(_LW("The QoS message_queue notification driver "
-        registry.provide(_get_qos_policy_cb, resources.QOS_POLICY)
+                        "has been ignored, since rpc push is implemented "
+                        "for any QoS driver that requests it."))
 
     def get_description(self):
         return "Message queue updates"
 
     def create_policy(self, context, policy):
-        #No need to update agents on create
         pass
 
     def update_policy(self, context, policy):
-        self.notification_api.push(context, [policy], events.UPDATED)
+        pass
 
     def delete_policy(self, context, policy):
-        self.notification_api.push(context, [policy], events.DELETED)
+        pass

neutron/services/qos/qos_consts.py

@@ -23,6 +23,8 @@ VALID_RULE_TYPES = [RULE_TYPE_BANDWIDTH_LIMIT,
 
 QOS_POLICY_ID = 'qos_policy_id'
 
+QOS_PLUGIN = 'qos_plugin'
+
 # NOTE(slaweq): Value used to calculate burst value for egress bandwidth limit
 # if burst is not given by user. In such case burst value will be calculated
 # as 80% of bw_limit to ensure that at least limits for TCP traffic will work

neutron/services/qos/qos_plugin.py

@@ -13,6 +13,8 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from oslo_log import log
+
 from neutron.common import exceptions as n_exc
 from neutron.db import api as db_api
 from neutron.db import db_base_plugin_common
@@ -20,9 +22,12 @@ from neutron.extensions import qos
 from neutron.objects import base as base_obj
 from neutron.objects.qos import policy as policy_object
 from neutron.objects.qos import rule_type as rule_type_object
+from neutron.services.qos.drivers import manager
 from neutron.services.qos.notification_drivers import manager as driver_mgr
 from neutron.services.qos import qos_consts
 
+LOG = log.getLogger(__name__)
+
 
 class QoSPlugin(qos.QoSPluginBase):
     """Implementation of the Neutron QoS Service Plugin.
@@ -38,9 +43,14 @@ class QoSPlugin(qos.QoSPluginBase):
 
     def __init__(self):
         super(QoSPlugin, self).__init__()
+
+        # TODO(mangelajo): remove notification_driver_manager in Pike
         self.notification_driver_manager = (
             driver_mgr.QosServiceNotificationDriverManager())
 
+        self.driver_manager = manager.QosServiceDriverManager(enable_rpc=(
+                self.notification_driver_manager.has_message_queue_driver))
+
     @db_base_plugin_common.convert_result_to_dict
     def create_policy(self, context, policy):
         """Create a QoS policy.
@@ -61,7 +71,12 @@ class QoSPlugin(qos.QoSPluginBase):
 
         policy_obj = policy_object.QosPolicy(context, **policy['policy'])
         policy_obj.create()
+
+        self.driver_manager.call('create_policy', context, policy_obj)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.create_policy(context, policy_obj)
+
         return policy_obj
 
     @db_base_plugin_common.convert_result_to_dict
@@ -81,7 +96,12 @@ class QoSPlugin(qos.QoSPluginBase):
         policy_obj = policy_object.QosPolicy(context, id=policy_id)
         policy_obj.update_fields(policy_data, reset_changes=True)
         policy_obj.update()
+
+        self.driver_manager.call('update_policy', context, policy_obj)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.update_policy(context, policy_obj)
+
         return policy_obj
 
     def delete_policy(self, context, policy_id):
@@ -97,6 +117,10 @@ class QoSPlugin(qos.QoSPluginBase):
         policy = policy_object.QosPolicy(context)
         policy.id = policy_id
         policy.delete()
+
+        self.driver_manager.call('delete_policy', context, policy)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.delete_policy(context, policy)
 
     def _get_policy_obj(self, context, policy_id):
@@ -155,6 +179,10 @@ class QoSPlugin(qos.QoSPluginBase):
             filters = {}
         return rule_type_object.QosRuleType.get_objects(**filters)
 
+    @property
+    def supported_rule_types(self):
+        return self.driver_manager.supported_rule_types
+
     @db_base_plugin_common.convert_result_to_dict
     def create_policy_rule(self, context, rule_cls, policy_id, rule_data):
         """Create a QoS policy rule.
@@ -179,7 +207,12 @@ class QoSPlugin(qos.QoSPluginBase):
             rule = rule_cls(context, qos_policy_id=policy_id, **rule_data)
             rule.create()
             policy.reload_rules()
+
+        self.driver_manager.call('update_policy', context, policy)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.update_policy(context, policy)
+
         return rule
 
     @db_base_plugin_common.convert_result_to_dict
@@ -212,7 +245,12 @@ class QoSPlugin(qos.QoSPluginBase):
             rule.update_fields(rule_data, reset_changes=True)
             rule.update()
             policy.reload_rules()
+
+        self.driver_manager.call('update_policy', context, policy)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.update_policy(context, policy)
+
         return rule
 
     def delete_policy_rule(self, context, rule_cls, rule_id, policy_id):
@@ -235,6 +273,10 @@ class QoSPlugin(qos.QoSPluginBase):
             rule = policy.get_rule_by_id(rule_id)
             rule.delete()
             policy.reload_rules()
+
+        self.driver_manager.call('update_policy', context, policy)
+
+        #TODO(majopela): remove notification_driver_manager call in Pike
         self.notification_driver_manager.update_policy(context, policy)
 
     @db_base_plugin_common.filter_fields

neutron/tests/base.py

@@ -42,6 +42,7 @@ import testtools
 from neutron._i18n import _
 from neutron.agent.linux import external_process
 from neutron.api.rpc.callbacks.consumer import registry as rpc_consumer_reg
+from neutron.api.rpc.callbacks.producer import registry as rpc_producer_reg
 from neutron.callbacks import manager as registry_manager
 from neutron.callbacks import registry
 from neutron.common import config
@@ -298,6 +299,7 @@ class BaseTestCase(DietTestCase):
         self.addCleanup(resource_registry.unregister_all_resources)
         self.addCleanup(db_api.sqla_remove_all)
         self.addCleanup(rpc_consumer_reg.clear)
+        self.addCleanup(rpc_producer_reg.clear)
 
     def get_new_temp_dir(self):
         """Create a new temporary directory.

neutron/tests/fullstack/test_qos.py

@@ -30,8 +30,7 @@ from neutron.tests.unit import testlib_api
 
 from neutron.plugins.ml2.drivers.linuxbridge.agent import \
     linuxbridge_neutron_agent as linuxbridge_agent
-from neutron.plugins.ml2.drivers.openvswitch.mech_driver import \
+from neutron.services.qos.drivers.openvswitch import driver as ovs_drv
-    mech_openvswitch as mech_ovs
 
 
 load_tests = testlib_api.module_load_tests
@@ -351,6 +350,5 @@ class TestQoSWithL2Population(base.BaseFullStackTestCase):
     def test_supported_qos_rule_types(self):
         res = self.client.list_qos_rule_types()
         rule_types = {t['type'] for t in res['rule_types']}
-        expected_rules = (
+        expected_rules = set(ovs_drv.SUPPORTED_RULES)
-            set(mech_ovs.OpenvswitchMechanismDriver.supported_qos_rule_types))
         self.assertEqual(expected_rules, rule_types)

neutron/tests/unit/plugins/ml2/test_plugin.py

@@ -58,7 +58,6 @@ from neutron.plugins.ml2 import managers
 from neutron.plugins.ml2 import models
 from neutron.plugins.ml2 import plugin as ml2_plugin
 from neutron.services.l3_router import l3_router_plugin
-from neutron.services.qos import qos_consts
 from neutron.services.revisions import revision_plugin
 from neutron.services.segments import db as segments_plugin_db
 from neutron.services.segments import plugin as segments_plugin
@@ -160,54 +159,6 @@ class TestMl2BulkToggleWithoutBulkless(Ml2PluginV2TestCase):
         self.assertFalse(self._skip_native_bulk)
 
 
-class TestMl2SupportedQosRuleTypes(Ml2PluginV2TestCase):
-
-    def test_empty_driver_list(self, *mocks):
-        mech_drivers_mock = mock.PropertyMock(return_value=[])
-        with mock.patch.object(self.driver.mechanism_manager,
-                               'ordered_mech_drivers',
-                               new_callable=mech_drivers_mock):
-            self.assertEqual(
-                [], self.driver.mechanism_manager.supported_qos_rule_types)
-
-    def test_no_rule_types_in_common(self):
-        self.assertEqual(
-            [], self.driver.mechanism_manager.supported_qos_rule_types)
-
-    @mock.patch.object(mech_logger.LoggerMechanismDriver,
-                       'supported_qos_rule_types',
-                       new_callable=mock.PropertyMock,
-                       create=True)
-    @mock.patch.object(mech_test.TestMechanismDriver,
-                       'supported_qos_rule_types',
-                       new_callable=mock.PropertyMock,
-                       create=True)
-    def test_rule_type_in_common(self, *mocks):
-        # make sure both plugins have the same supported qos rule types
-        for mock_ in mocks:
-            mock_.return_value = qos_consts.VALID_RULE_TYPES
-        for rule in qos_consts.VALID_RULE_TYPES:
-            self.assertIn(
-                rule,
-                self.driver.mechanism_manager.supported_qos_rule_types)
-
-    @mock.patch.object(mech_test.TestMechanismDriver,
-                       'supported_qos_rule_types',
-                       new_callable=mock.PropertyMock,
-                       return_value=qos_consts.VALID_RULE_TYPES,
-                       create=True)
-    @mock.patch.object(mech_logger.LoggerMechanismDriver,
-                       '_supports_port_binding',
-                       new_callable=mock.PropertyMock,
-                       return_value=False)
-    def test_rule_types_with_driver_that_does_not_implement_binding(self,
-                                                                    *mocks):
-        for rule in qos_consts.VALID_RULE_TYPES:
-            self.assertIn(
-                rule,
-                self.driver.mechanism_manager.supported_qos_rule_types)
-
-
 class TestMl2BasicGet(test_plugin.TestBasicGet,
                       Ml2PluginV2TestCase):
     pass

neutron/tests/unit/services/qos/drivers/test_manager.py

@@ -0,0 +1,94 @@
+#    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 oslo_config import cfg
+
+from neutron.conf.services import qos_driver_manager as notif_driver_mgr_config
+from neutron.services.qos.drivers import base as qos_driver_base
+from neutron.services.qos.drivers import manager as driver_mgr
+from neutron.services.qos import qos_consts
+from neutron.tests.unit.services.qos import base
+
+
+class TestQosDriversManagerBase(base.BaseQosTestCase):
+
+    def setUp(self):
+        super(TestQosDriversManagerBase, self).setUp()
+        self.config_parse()
+        self.setup_coreplugin(load_plugins=False)
+        config = cfg.ConfigOpts()
+        notif_driver_mgr_config.register_qos_plugin_opts(config)
+
+    @staticmethod
+    def _create_manager_with_drivers(drivers_details):
+        for name, driver_details in drivers_details.items():
+
+            class QoSDriver(qos_driver_base.DriverBase):
+                @property
+                def is_loaded(self):
+                    return driver_details['is_loaded']
+
+            # the new ad-hoc driver will register on the QOS_PLUGIN registry
+            QoSDriver(name,
+                      driver_details.get('vif_types', []),
+                      driver_details.get('vnic_types', []),
+                      driver_details.get('rules', []))
+
+        return driver_mgr.QosServiceDriverManager()
+
+
+class TestQosDriversManagerMulti(TestQosDriversManagerBase):
+    """Test calls happen to all drivers"""
+    def test_driver_manager_empty_with_no_drivers(self):
+        driver_manager = self._create_manager_with_drivers({})
+        self.assertEqual(len(driver_manager._drivers), 0)
+
+    def test_driver_manager_empty_with_no_loaded_drivers(self):
+        driver_manager = self._create_manager_with_drivers(
+            {'driver-A': {'is_loaded': False}})
+        self.assertEqual(len(driver_manager._drivers), 0)
+
+    def test_driver_manager_with_one_loaded_driver(self):
+        driver_manager = self._create_manager_with_drivers(
+            {'driver-A': {'is_loaded': True}})
+        self.assertEqual(len(driver_manager._drivers), 1)
+
+    def test_driver_manager_with_two_loaded_drivers(self):
+        driver_manager = self._create_manager_with_drivers(
+            {'driver-A': {'is_loaded': True},
+             'driver-B': {'is_loaded': True}})
+        self.assertEqual(len(driver_manager._drivers), 2)
+
+
+class TestQosDriversManagerRules(TestQosDriversManagerBase):
+    """Test supported rules"""
+    def test_available_rules_one_in_common(self):
+        driver_manager = self._create_manager_with_drivers(
+            {'driver-A': {'is_loaded': True,
+                          'rules': [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
+                                    qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]},
+             'driver-B': {'is_loaded': True,
+                          'rules': [qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH,
+                                    qos_consts.RULE_TYPE_DSCP_MARKING]}
+             })
+        self.assertEqual(driver_manager.supported_rule_types,
+                         set([qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]))
+
+    def test_available_rules_no_rule_in_common(self):
+        driver_manager = self._create_manager_with_drivers(
+            {'driver-A': {'is_loaded': True,
+                          'rules': [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT]},
+             'driver-B': {'is_loaded': True,
+                          'rules': [qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH,
+                                    qos_consts.RULE_TYPE_DSCP_MARKING]}
+             })
+        self.assertEqual(driver_manager.supported_rule_types, set([]))

neutron/tests/unit/services/qos/notification_drivers/test_manager.py

@@ -14,7 +14,6 @@ import mock
 from oslo_config import cfg
 from oslo_utils import uuidutils
 
-from neutron.api.rpc.callbacks import events
 from neutron.conf.services import qos_driver_manager as driver_mgr_config
 from neutron import context
 from neutron.objects.qos import policy as policy_object
@@ -55,37 +54,6 @@ class TestQosDriversManagerBase(base.BaseQosTestCase):
         self.kwargs = {'context': ctxt}
 
 
-class TestQosDriversManager(TestQosDriversManagerBase):
-
-    def setUp(self):
-        super(TestQosDriversManager, self).setUp()
-        #TODO(Qos): Fix this unittest to test manager and not message_queue
-        #           notification driver
-        rpc_api_cls = mock.patch('neutron.api.rpc.handlers.resources_rpc'
-                                 '.ResourcesPushRpcApi').start()
-        self.rpc_api = rpc_api_cls.return_value
-        self.driver_manager = driver_mgr.QosServiceNotificationDriverManager()
-
-    def _validate_registry_params(self, event_type, policy):
-        self.rpc_api.push.assert_called_with(self.context, [policy],
-                                             event_type)
-
-    def test_create_policy_default_configuration(self):
-        #RPC driver should be loaded by default
-        self.driver_manager.create_policy(self.context, self.policy)
-        self.assertFalse(self.rpc_api.push.called)
-
-    def test_update_policy_default_configuration(self):
-        #RPC driver should be loaded by default
-        self.driver_manager.update_policy(self.context, self.policy)
-        self._validate_registry_params(events.UPDATED, self.policy)
-
-    def test_delete_policy_default_configuration(self):
-        #RPC driver should be loaded by default
-        self.driver_manager.delete_policy(self.context, self.policy)
-        self._validate_registry_params(events.DELETED, self.policy)
-
-
 class TestQosDriversManagerMulti(TestQosDriversManagerBase):
 
     def _test_multi_drivers_configuration_op(self, op):

neutron/tests/unit/services/qos/notification_drivers/test_message_queue.py

@@ -1,69 +0,0 @@
-#    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 oslo_utils import uuidutils
-
-from neutron.api.rpc.callbacks import events
-from neutron import context
-from neutron.objects.qos import policy as policy_object
-from neutron.objects.qos import rule as rule_object
-from neutron.services.qos.notification_drivers import message_queue
-from neutron.tests.unit.services.qos import base
-
-DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
-
-
-class TestQosRpcNotificationDriver(base.BaseQosTestCase):
-
-    def setUp(self):
-        super(TestQosRpcNotificationDriver, self).setUp()
-        rpc_api_cls = mock.patch('neutron.api.rpc.handlers.resources_rpc'
-                                 '.ResourcesPushRpcApi').start()
-        self.rpc_api = rpc_api_cls.return_value
-        self.driver = message_queue.RpcQosServiceNotificationDriver()
-        policy_id = uuidutils.generate_uuid()
-        self.policy_data = {'policy': {
-                            'id': policy_id,
-                            'project_id': uuidutils.generate_uuid(),
-                            'name': 'testi-policy',
-                            'description': 'test policyi description',
-                            'shared': True}}
-
-        self.rule_data = {'bandwidth_limit_rule': {
-                            'id': policy_id,
-                            'max_kbps': 100,
-                            'max_burst_kbps': 150}}
-
-        self.context = context.get_admin_context()
-        self.policy = policy_object.QosPolicy(self.context,
-                        **self.policy_data['policy'])
-
-        self.rule = rule_object.QosBandwidthLimitRule(
-                                self.context,
-                                **self.rule_data['bandwidth_limit_rule'])
-
-    def _validate_push_params(self, event_type, policy):
-        self.rpc_api.push.assert_called_once_with(self.context, [policy],
-                                                  event_type)
-
-    def test_create_policy(self):
-        self.driver.create_policy(self.context, self.policy)
-        self.assertFalse(self.rpc_api.push.called)
-
-    def test_update_policy(self):
-        self.driver.update_policy(self.context, self.policy)
-        self._validate_push_params(events.UPDATED, self.policy)
-
-    def test_delete_policy(self):
-        self.driver.delete_policy(self.context, self.policy)
-        self._validate_push_params(events.DELETED, self.policy)

neutron/tests/unit/services/qos/test_qos_plugin.py

@@ -51,7 +51,13 @@ class TestQosPlugin(base.BaseQosTestCase):
 
         manager.init()
         self.qos_plugin = directory.get_plugin(constants.QOS)
+
+        #TODO(mangelajo): Remove notification_driver_manager mock in Pike
         self.qos_plugin.notification_driver_manager = mock.Mock()
+        self.qos_plugin.driver_manager = mock.Mock()
+
+        self.rpc_push = mock.patch('neutron.api.rpc.handlers.resources_rpc'
+                                   '.ResourcesPushRpcApi.push').start()
 
         self.ctxt = context.Context('fake_user', 'fake_tenant')
         mock.patch.object(self.ctxt.session, 'refresh').start()
@@ -80,19 +86,27 @@ class TestQosPlugin(base.BaseQosTestCase):
         self.dscp_rule = rule_object.QosDscpMarkingRule(
             self.ctxt, **self.rule_data['dscp_marking_rule'])
 
-    def _validate_notif_driver_params(self, method_name):
+    def _validate_driver_params(self, method_name):
         method = getattr(self.qos_plugin.notification_driver_manager,
                          method_name)
         self.assertTrue(method.called)
         self.assertIsInstance(
             method.call_args[0][1], policy_object.QosPolicy)
 
+        self.assertTrue(self.qos_plugin.driver_manager.call.called)
+        self.assertEqual(self.qos_plugin.driver_manager.call.call_args[0][0],
+                         method_name)
+        self.assertIsInstance(
+            self.qos_plugin.driver_manager.call.call_args[0][2],
+            policy_object.QosPolicy
+        )
+
     @mock.patch(
         'neutron.objects.rbac_db.RbacNeutronDbObjectMixin'
         '.create_rbac_policy')
     def test_add_policy(self, *mocks):
         self.qos_plugin.create_policy(self.ctxt, self.policy_data)
-        self._validate_notif_driver_params('create_policy')
+        self._validate_driver_params('create_policy')
 
     def test_add_policy_with_extra_tenant_keyword(self, *mocks):
         policy_id = uuidutils.generate_uuid()
@@ -124,19 +138,19 @@ class TestQosPlugin(base.BaseQosTestCase):
             policy_object.QosPolicy, self.policy_data['policy'])
         self.qos_plugin.update_policy(
             self.ctxt, self.policy.id, {'policy': fields})
-        self._validate_notif_driver_params('update_policy')
+        self._validate_driver_params('update_policy')
 
     @mock.patch('neutron.objects.db.api.get_object', return_value=None)
     def test_delete_policy(self, *mocks):
         self.qos_plugin.delete_policy(self.ctxt, self.policy.id)
-        self._validate_notif_driver_params('delete_policy')
+        self._validate_driver_params('delete_policy')
 
     def test_create_policy_rule(self):
         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
                         return_value=self.policy):
             self.qos_plugin.create_policy_bandwidth_limit_rule(
                 self.ctxt, self.policy.id, self.rule_data)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_update_policy_rule(self):
         _policy = policy_object.QosPolicy(
@@ -146,7 +160,7 @@ class TestQosPlugin(base.BaseQosTestCase):
             setattr(_policy, "rules", [self.rule])
             self.qos_plugin.update_policy_bandwidth_limit_rule(
                 self.ctxt, self.rule.id, self.policy.id, self.rule_data)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_update_policy_rule_bad_policy(self):
         _policy = policy_object.QosPolicy(
@@ -168,7 +182,7 @@ class TestQosPlugin(base.BaseQosTestCase):
             setattr(_policy, "rules", [self.rule])
             self.qos_plugin.delete_policy_bandwidth_limit_rule(
                         self.ctxt, self.rule.id, _policy.id)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_delete_policy_rule_bad_policy(self):
         _policy = policy_object.QosPolicy(
@@ -250,7 +264,7 @@ class TestQosPlugin(base.BaseQosTestCase):
             setattr(_policy, "rules", [self.dscp_rule])
             self.qos_plugin.create_policy_dscp_marking_rule(
                 self.ctxt, self.policy.id, self.rule_data)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_update_policy_dscp_marking_rule(self):
         _policy = policy_object.QosPolicy(
@@ -260,7 +274,7 @@ class TestQosPlugin(base.BaseQosTestCase):
             setattr(_policy, "rules", [self.dscp_rule])
             self.qos_plugin.update_policy_dscp_marking_rule(
                 self.ctxt, self.dscp_rule.id, self.policy.id, self.rule_data)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_delete_policy_dscp_marking_rule(self):
         _policy = policy_object.QosPolicy(
@@ -270,7 +284,7 @@ class TestQosPlugin(base.BaseQosTestCase):
             setattr(_policy, "rules", [self.dscp_rule])
             self.qos_plugin.delete_policy_dscp_marking_rule(
                 self.ctxt, self.dscp_rule.id, self.policy.id)
-            self._validate_notif_driver_params('update_policy')
+            self._validate_driver_params('update_policy')
 
     def test_get_policy_dscp_marking_rules(self):
         with mock.patch('neutron.objects.qos.policy.QosPolicy.get_object',
@@ -436,7 +450,7 @@ class TestQosPlugin(base.BaseQosTestCase):
                             mock_manager.mock_calls.index(notify_mock_call))
 
     @mock.patch('neutron.objects.qos.policy.QosPolicy')
-    def test_rule_notification_ordering(self, qos_policy_mock):
+    def test_rule_notification_and_driver_ordering(self, qos_policy_mock):
         rule_cls_mock = mock.Mock()
         rule_cls_mock.rule_type = 'fake'
 
@@ -448,6 +462,8 @@ class TestQosPlugin(base.BaseQosTestCase):
                         'delete': [self.ctxt, rule_cls_mock,
                                    self.rule.id, self.policy.id]}
 
+        # TODO(mangelajo): Remove notification_driver_manager checks in Pike
+        #                  and rename this test
         self.qos_plugin.notification_driver_manager = mock.Mock()
 
         mock_manager = mock.Mock()
@@ -455,6 +471,7 @@ class TestQosPlugin(base.BaseQosTestCase):
         mock_manager.attach_mock(rule_cls_mock, 'RuleCls')
         mock_manager.attach_mock(self.qos_plugin.notification_driver_manager,
                                  'notification_driver')
+        mock_manager.attach_mock(self.qos_plugin.driver_manager, 'driver')
 
         for action, arguments in rule_actions.items():
             mock_manager.reset_mock()
@@ -470,6 +487,9 @@ class TestQosPlugin(base.BaseQosTestCase):
             notify_mock_call = mock.call.notification_driver.update_policy(
                     self.ctxt, mock.ANY)
 
+            driver_mock_call = mock.call.driver.call('update_policy',
+                                                     self.ctxt, mock.ANY)
+
             if rule_mock_call in mock_manager.mock_calls:
                 action_index = mock_manager.mock_calls.index(rule_mock_call)
             else:
@@ -478,3 +498,6 @@ class TestQosPlugin(base.BaseQosTestCase):
 
             self.assertTrue(
                 action_index < mock_manager.mock_calls.index(notify_mock_call))
+
+            self.assertTrue(
+                action_index < mock_manager.mock_calls.index(driver_mock_call))

neutron/tests/unit/services/revisions/test_revision_plugin.py

@@ -13,7 +13,6 @@
 #    under the License.
 #
 
-import mock
 import netaddr
 from neutron_lib import constants
 from neutron_lib.plugins import directory
@@ -43,8 +42,6 @@ class TestRevisionPlugin(test_plugin.Ml2PluginV2TestCase):
         config.cfg.CONF.set_override('extension_drivers',
                                      self._extension_drivers,
                                      group='ml2')
-        mock.patch('neutron.services.qos.notification_drivers.message_queue'
-                   '.RpcQosServiceNotificationDriver').start()
         super(TestRevisionPlugin, self).setUp()
         self.cp = directory.get_plugin()
         self.l3p = directory.get_plugin(constants.L3)

releasenotes/notes/qos-drivers-refactor-16ece9984958f8a4.yaml

@@ -0,0 +1,12 @@
+---
+features:
+  - The QoS driver architecture has been refactored to overcome several
+    previous limitations, the main one was the coupling of QoS details
+    into the mechanism drivers, and the next one was the need of
+    configuration knobs to enable each specific notification driver,
+    that will be handled automatically from now on.
+
+deprecations:
+  - |
+    notification_drivers from [qos] section has been deprecated. It will be
+    removed in a future release.