Adds QoS support

This change adds QoS support for the neutron Hyper-V agent,
leveraging the QoS Neutron extension.

Depends-On: I706368bfcaece380e1357e0c504fd3b9553ba49c
Depends-On: I1dd810e238389456efd4048ebb0bf50f5ea2d237

Implements: blueprint hyperv-neutron-qos
Change-Id: I8f5adfdac7885f98508c80378d1fa467a6d4cf94

hyperv/neutron/config.py

@@ -56,6 +56,9 @@ HYPERV_AGENT_OPTS = [
               default='169.254.169.254',
               help=_('Specifies the address which will serve the metadata for'
                       ' the instance.')),
+    cfg.BoolOpt('enable_qos_extension',
+                default=False,
+                help=_('Enables the QoS extension.')),
 ]
 
 NVGRE_OPTS = [

hyperv/neutron/hyperv_neutron_agent.py

@@ -338,6 +338,8 @@ networking-plugin-hyperv_agent.html
                                  device_details['physical_network'],
                                  device_details['segmentation_id'],
                                  device_details['admin_state_up'])
+            if CONF.AGENT.enable_qos_extension:
+                self._qos_ext.handle_port(self.context, device_details)
 
             LOG.debug("Updating cached port %s status as UP.", port_id)
             self._update_port_status_cache(device, device_bound=True)

hyperv/neutron/l2_agent.py

@@ -18,6 +18,7 @@ import platform
 import sys
 
 from neutron.agent.common import config
+from neutron.agent.l2.extensions import qos as qos_extension
 from neutron.agent import rpc as agent_rpc
 from neutron.agent import securitygroups_rpc as sg_rpc
 from neutron.common import config as common_config
@@ -36,6 +37,7 @@ from hyperv.neutron import hyperv_neutron_agent
 
 LOG = logging.getLogger(__name__)
 CONF = cfg.CONF
+CONF.import_group('AGENT', 'hyperv.neutron.config')
 
 
 class HyperVSecurityAgent(sg_rpc.SecurityGroupAgentRpc):
@@ -132,7 +134,15 @@ class HyperVNeutronAgent(hyperv_neutron_agent.HyperVNeutronAgentMixin):
 
         self.connection = agent_rpc.create_consumers(self.endpoints,
                                                      self.topic,
-                                                     consumers)
+                                                     consumers,
+                                                     start_listening=False)
+
+        if CONF.AGENT.enable_qos_extension:
+            self._qos_ext = qos_extension.QosAgentExtension()
+            self._qos_ext.consume_api(self)
+            self._qos_ext.initialize(self.connection, 'hyperv')
+
+        self.connection.consume_in_threads()
 
         self.client = n_rpc.get_client(self.target)
         report_interval = CONF.AGENT.report_interval

hyperv/neutron/qos/qos_driver.py

@@ -0,0 +1,90 @@
+# Copyright 2017 Cloudbase Solutions Srl
+#
+# 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.agent.l2.extensions import qos
+from neutron.services.qos import qos_consts
+from os_win.utils.network import networkutils
+from oslo_log import log as logging
+
+from hyperv.common.i18n import _LI, _LW  # noqa
+
+LOG = logging.getLogger(__name__)
+
+
+class QosHyperVAgentDriver(qos.QosAgentDriver):
+
+    _SUPPORTED_QOS_RULES = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
+                            qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH]
+
+    def initialize(self):
+        self._utils = networkutils.NetworkUtils()
+
+    def create(self, port, qos_policy):
+        """Apply QoS rules on port for the first time.
+
+        :param port: port object.
+        :param qos_policy: the QoS policy to be applied on port.
+        """
+        LOG.info(_LI("Setting QoS policy %(qos_policy)s on "
+                     "port %(port)s"),
+                 dict(qos_policy=qos_policy,
+                      port=port))
+
+        policy_data = self._get_policy_values(qos_policy)
+        self._utils.set_port_qos_rule(port["port_id"], policy_data)
+
+    def update(self, port, qos_policy):
+        """Apply QoS rules on port.
+
+        :param port: port object.
+        :param qos_policy: the QoS policy to be applied on port.
+        """
+        LOG.info(_LI("Updating QoS policy %(qos_policy)s on "
+                     "port %(port)s"),
+                 dict(qos_policy=qos_policy,
+                      port=port))
+
+        policy_data = self._get_policy_values(qos_policy)
+        self._utils.set_port_qos_rule(port["port_id"], policy_data)
+
+    def delete(self, port, qos_policy=None):
+        """Remove QoS rules from port.
+
+        :param port: port object.
+        :param qos_policy: the QoS policy to be removed from port.
+        """
+        LOG.info(_LI("Deleting QoS policy %(qos_policy)s on "
+                     "port %(port)s"),
+                 dict(qos_policy=qos_policy,
+                      port=port))
+
+        self._utils.remove_port_qos_rule(port["port_id"])
+
+    def _get_policy_values(self, qos_policy):
+        result = {}
+        for qos_rule in qos_policy.rules:
+            if qos_rule.rule_type not in self._SUPPORTED_QOS_RULES:
+                LOG.warning(_LW("Unsupported QoS rule: %(qos_rule)s"),
+                            dict(qos_rule=qos_rule))
+                continue
+            result['min_kbps'] = getattr(qos_rule, 'min_kbps',
+                                         result.get('min_kbps'))
+            result['max_kbps'] = getattr(qos_rule, 'max_kbps',
+                                         result.get('max_kbps'))
+            result['max_burst_kbps'] = getattr(qos_rule, 'max_burst_kbps',
+                                               result.get('max_burst_kbps'))
+
+        return result

hyperv/tests/unit/neutron/qos/test_qos_driver.py

@@ -0,0 +1,76 @@
+# Copyright 2017 Cloudbase Solutions Srl
+# 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.
+
+"""
+Unit tests for Windows Hyper-V QoS Driver.
+"""
+
+import mock
+from neutron.services.qos import qos_consts
+
+from hyperv.neutron.qos import qos_driver
+from hyperv.tests import base
+
+
+class TestQosHyperVAgentDriver(base.BaseTestCase):
+    @mock.patch.object(qos_driver.QosHyperVAgentDriver, '__init__',
+                       lambda *args, **kwargs: None)
+    def setUp(self):
+        super(TestQosHyperVAgentDriver, self).setUp()
+        self.driver = qos_driver.QosHyperVAgentDriver()
+        self.driver._utils = mock.Mock()
+
+    @mock.patch.object(qos_driver, 'networkutils')
+    def test_initialize(self, mock_networkutils):
+        self.driver.initialize()
+        mock_networkutils.NetworkUtils.assert_called_once_with()
+
+    @mock.patch.object(qos_driver.QosHyperVAgentDriver, '_get_policy_values')
+    def test_create(self, mock_get_policy_values):
+        self.driver.create({'port_id': mock.sentinel.port_id},
+                           mock.sentinel.qos_policy)
+        mock_get_policy_values.assert_called_once_with(
+            mock.sentinel.qos_policy)
+        self.driver._utils.set_port_qos_rule.assert_called_once_with(
+            mock.sentinel.port_id, mock_get_policy_values.return_value)
+
+    @mock.patch.object(qos_driver.QosHyperVAgentDriver, '_get_policy_values')
+    def test_update(self, mock_get_policy_values):
+        self.driver.update({'port_id': mock.sentinel.port_id},
+                           mock.sentinel.qos_policy)
+        mock_get_policy_values.assert_called_once_with(
+            mock.sentinel.qos_policy)
+        self.driver._utils.set_port_qos_rule.assert_called_once_with(
+            mock.sentinel.port_id, mock_get_policy_values.return_value)
+
+    def test_delete(self):
+        self.driver.delete({'port_id': mock.sentinel.port_id})
+        self.driver._utils.remove_port_qos_rule.assert_called_once_with(
+            mock.sentinel.port_id)
+
+    def test_get_policy_values(self):
+        qos_rule_0 = mock.Mock(spec=['min_kbps', 'rule_type'])
+        qos_rule_0.rule_type = qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH
+        qos_rule_1 = mock.Mock(spec=['max_kbps', 'max_burst_kbps',
+                                     'rule_type'])
+        qos_rule_1.rule_type = qos_consts.RULE_TYPE_BANDWIDTH_LIMIT
+        qos_policy = mock.Mock(rules=[qos_rule_0, qos_rule_1])
+
+        expected_val = dict(min_kbps=qos_rule_0.min_kbps,
+                            max_kbps=qos_rule_1.max_kbps,
+                            max_burst_kbps=qos_rule_1.max_burst_kbps)
+        policy_val = self.driver._get_policy_values(qos_policy)
+
+        self.assertEqual(expected_val, policy_val)

hyperv/tests/unit/neutron/test_hyperv_neutron_agent.py

@@ -42,6 +42,7 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
         self.addCleanup(utilsfactory_patcher.stop)
 
         self.agent = hyperv_neutron_agent.HyperVNeutronAgentMixin()
+        self.agent._qos_ext = mock.MagicMock()
         self.agent.plugin_rpc = mock.Mock()
         self.agent._metricsutils = mock.MagicMock()
         self.agent._utils = mock.MagicMock()
@@ -480,9 +481,11 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
                                 mock_treat_vif_port):
         self.agent._added_ports = set()
         details = self._get_fake_port_details()
-
+        CONF.AGENT.enable_qos_extension = True
         self.agent._process_added_port(details)
 
+        self.agent._qos_ext.handle_port.assert_called_once_with(
+            self.agent.context, details)
         mock_treat_vif_port.assert_called_once_with(
             mock.sentinel.port_id, mock.sentinel.network_id,
             mock.sentinel.network_type, mock.sentinel.physical_network,

hyperv/tests/unit/neutron/test_l2_agent.py

@@ -119,6 +119,7 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
             self.agent.context, {'start_flag': True})
         self.assertTrue(self.agent.agent_state['start_flag'])
 
+    @mock.patch.object(l2_agent.qos_extension, 'QosAgentExtension')
     @mock.patch.object(l2_agent.loopingcall, 'FixedIntervalLoopingCall')
     @mock.patch.object(l2_agent.n_rpc, 'get_client')
     @mock.patch.object(l2_agent, 'HyperVSecurityAgent')
@@ -127,9 +128,10 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
     @mock.patch.object(l2_agent, 'CONF')
     def test_setup_rpc(self, mock_CONF, mock_agent_rpc, mock_SGRpcApi,
                        mock_HyperVSecurityAgent, mock_get_client,
-                       mock_LoopingCall):
+                       mock_LoopingCall, mock_qos_ext):
         mock_CONF.NVGRE.enable_support = True
         mock_CONF.AGENT.report_interval = mock.sentinel.report_interval
+        mock_CONF.AGENT.enable_qos_extension = True
         self.agent._setup_rpc()
 
         self.assertEqual('hyperv_%s' % platform.node(), self.agent.agent_id)
@@ -152,7 +154,13 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
                      [constants.TUNNEL, topics.UPDATE],
                      [constants.LOOKUP, constants.UPDATE]]
         mock_agent_rpc.create_consumers.assert_called_once_with(
-            self.agent.endpoints, self.agent.topic, consumers)
+            self.agent.endpoints, self.agent.topic,
+            consumers, start_listening=False)
+        mock_qos_ext.return_value.consume_api.assert_called_once_with(
+            self.agent)
+        mock_qos_ext.return_value.initialize.assert_called_once_with(
+            self.agent.connection, 'hyperv')
+        self.agent.connection.consume_in_threads.assert_called_once_with()
         mock_LoopingCall.return_value.start.assert_called_once_with(
             interval=mock.sentinel.report_interval)
 

setup.cfg

@@ -29,6 +29,9 @@ packages =
 console_scripts =
     neutron-hyperv-agent = hyperv.neutron.l2_agent:main
 
+neutron.qos.agent_drivers =
+    hyperv = hyperv.neutron.qos.qos_driver:QosHyperVAgentDriver
+
 neutron.ml2.mechanism_drivers =
     hyperv = hyperv.neutron.ml2.mech_hyperv:HypervMechanismDriver