openstack/neutron
Code Issues Proposed changes

Merge remote-tracking branch 'origin/feature/qos' into merge-branch

Browse Source 
Change-Id: I1c1fd593235fda1cdd053980f50eff21ca9011b6
This commit is contained in:
Ihar Hrachyshka 2015-07-24 14:24:10 +02:00
commit fcc5d5bcf7
77 changed files with 3626 additions and 16 deletions

1
.gitreview

@ -2,3 +2,4 @@
host=review.openstack.org
port=29418
project=openstack/neutron.git
defaultbranch=feature/qos

2
doc/source/devref/index.rst

@ -48,8 +48,10 @@ Neutron Internals
plugin-api
db_layer
rpc_api
rpc_callbacks
layer3
l2_agents
quality_of_service
advanced_services
oslo-incubator
callbacks

256
doc/source/devref/quality_of_service.rst Normal file

@ -0,0 +1,256 @@
==================
Quality of Service
==================
Quality of Service advanced service is designed as a service plugin. The
service is decoupled from the rest of Neutron code on multiple levels (see
below).
QoS is the first service/api extension to extend core resources (ports,
networks) without using mixins inherited from plugins.
Details about the DB models, API extension, and use cases can be found here: `qos spec <http://specs.openstack.org/openstack/neutron-specs/specs/liberty/qos-api-extension.html>`_
.
Service side design
===================
* neutron.extensions.qos:
base extension + API controller definition.
* neutron.services.qos.qos_plugin:
QoSPlugin, service plugin that implements 'qos' extension, receiving and
handling API calls to create/modify policies and rules. It also handles core
plugin requests to associate ports and networks with a QoS policy.
* neutron.services.qos.drivers.qos_base:
the interface class for server-side QoS backend which will receive {create,
update, delete} events on any rule change.
* neutron.services.qos.drivers.rpc.mq_qos:
message queue based reference backend driver which provides messaging
notifications to any interested agent, using `RPC callbacks <rpc_callbacks.html>`_.
QoS resources
-------------
QoS design defines the following two conceptual resources to define QoS rules
for a port or a network:
* QoS policy
* QoS rule (type specific)
Each QoS policy contains zero or more QoS rules. A policy is then applied to a
network or a port, making all rules of the policy applied to the corresponding
Neutron resource (for a network, applying a policy means that the policy will
be applied to all ports that belong to it).
From database point of view, following objects are defined in schema:
* QosPolicy: directly maps to the conceptual policy resource.
* QosNetworkPolicyBinding, QosPortPolicyBinding: defines attachment between a
Neutron resource and a QoS policy.
* QosRule: defines common rule fields for all supported rule types.
* QosBandwidthLimitRule: defines rule fields that are specific to
bandwidth_limit type (the only type supported by the service as of time of
writing).
There is a one-to-one relationship between QosRule and type specific
Qos<type>Rule database objects. We represent the single object with two tables
to avoid duplication of common fields. (That introduces some complexity in
neutron objects for rule resources, but see below).
All database models are defined under:
* neutron.db.qos.models
There is a long history of passing database dictionaries directly into business
logic of Neutron. This path is not the one we wanted to take for QoS effort, so
we've also introduced a new objects middleware to encapsulate the database logic
from the rest of the Neutron code that works with QoS resources. For this, we've
adopted oslo.versionedobjects library and introduced a new NeutronObject class
that is a base for all other objects that will belong to the middle layer.
There is an expectation that Neutron will evolve into using objects for all
resources it handles, though that part is obviously out of scope for the QoS
effort.
Every NeutronObject supports the following operations:
* get_by_id: returns specific object that is represented by the id passed as an
argument.
* get_objects: returns all objects of the type, potentially with a filter
applied.
* create/update/delete: usual persistence operations.
Base object class is defined in:
* neutron.objects.base
For QoS, new neutron objects were implemented:
* QosPolicy: directly maps to the conceptual policy resource, as defined above.
* QosBandwidthLimitRule: class that represents the only rule type supported by
initial QoS design.
Those are defined in:
* neutron.objects.qos.policy
* neutron.objects.qos.rule
For QosPolicy neutron object, the following public methods were implemented:
* get_network_policy/get_port_policy: returns a policy object that is attached
to the corresponding Neutron resource.
* attach_network/attach_port: attach a policy to the corresponding Neutron
resource.
* detach_network/detach_port: detach a policy from the corresponding Neutron
resource.
In addition to the fields that belong to QoS policy database object itself,
synthetic fields were added to the object that represent lists of rules,
per-type, that belong to the policy. For example, to get a list of all
bandwidth_limit rules for a specific policy, a consumer of the object can just
access corresponding attribute via:
* policy.bandwidth_limit_rules
Implementation is done in a way that will allow adding a new rule list field
with little or no modifications in the policy object itself. This is achieved
by smart introspection of existing available rule object definitions and
automatic definition of those fields on the policy class.
Note that synthetic fields are lazily loaded, meaning there is no hit into
the database if the field is not inspected by consumers.
For Qos<type>Rule objects, an extendable approach was taken to allow easy
addition of objects for new rule types. To accomodate this, all the methods
that access the database were implemented in a base class called QosRule that
is then inherited into type-specific rule implementations that, ideally, only
define additional fields and some other minor things.
Note that the QosRule base class is not registered with oslo.versionedobjects
registry, because it's not expected that 'generic' rules should be
instantiated (and to enforce just that, the base rule class is marked as ABC).
QoS objects rely on some primitive database API functions that are added in:
* neutron.db.api
* neutron.db.qos.api
Callback changes
----------------
TODO(QoS): We're changing strategy here to not rely on AFTER_READ callbacks,
and foster discussion about how to do decouple core resource
extension in the community. So, update next phrase when that
happens.
To extend ports and networks with qos_policy_id field, AFTER_READ callback
event is introduced.
Note: a better mechanism is being built by @armax to make resource extensions
more explicit and under control. We will migrate to that better mechanism as
soon as it's available.
RPC communication
-----------------
Details on RPC communication implemented in reference backend driver are
discussed in `a separate page <rpc_callbacks.html>`_.
One thing that should be mentioned here explicitly is that RPC callback
endpoints communicate using real versioned objects (as defined by serialization
for oslo.versionedobjects library), not vague json dictionaries. Meaning,
oslo.versionedobjects are on the wire and not just used internally inside a
component.
There is expectation that after RPC callbacks are introduced in Neutron, we
will be able to migrate propagation from server to agents for other resources
(f.e. security groups) to the new mechanism. This will need to wait until those
resources get proper NeutronObject implementations.
Agent side design
=================
To facilitate code reusability between agents and agent extensions without
patching the agent code itself, agent extensions were introduced. They can be
especially interesting to third parties that don't want to maintain their code
in Neutron tree.
Extensions are meant to receive basic events like port update or delete, and do
whatever they need with it.
* neutron.agent.l2.agent_extension:
extension interface definition.
* neutron.agent.l2.agent_extensions_manager:
manager that allows to register multiple extensions, and pass events down to
all enabled extensions.
* neutron.agent.l2.extensions.qos_agent:
defines QoSAgentExtension that is also pluggable using QoSAgentDriver
implementations that are specific to agent backends being used.
* neutron.agent.l2.l2_agent:
provides the API entry point for process_{network,subnet,port}_extension,
and holds an agent extension manager inside.
TODO(QoS): clarify what this is for, I don't follow a bit.
ML2
---
TODO(QoS): there is work ongoing that will need to be reflected here.
Agent backends
--------------
TODO(QoS): this section needs rework.
Open vSwitch
* neutron.plugins.ml2.drivers.openvswitch.agent.extension_drivers.qos_driver
This module implements the QoSAgentDriver interface used by the
QosAgentExtension.
* neutron.agent.common.ovs_lib
* neutron.agent.ovsdb.api
* neutron.agent.ovsdb.impl_idl
* neutron.agent.ovsdb.impl_vsctl
* neutron.agent.ovsdb.native.commands
SR-IOV
Configuration
=============
TODO(QoS)
Testing strategy
================
Neutron objects
---------------
Base unit test classes to validate neutron objects were implemented in a way
that allows code reuse when introducing a new object type.
There are two test classes that are utilized for that:
* BaseObjectIfaceTestCase: class to validate basic object operations (mostly
CRUD) with database layer isolated.
* BaseDbObjectTestCase: class to validate the same operations with models in
place and database layer unmocked.
Every new object implemented on top of one of those classes is expected to
either inherit existing test cases as is, or reimplement it, if it makes sense
in terms of how those objects are implemented. Specific test classes can
obviously extend the set of test cases as they see needed (f.e. you need to
define new test cases for those additional methods that you may add to your
object implementations on top of base semantics common to all neutron objects).

229
doc/source/devref/rpc_callbacks.rst Normal file

@ -0,0 +1,229 @@
=================================
Neutron Messaging Callback System
=================================
Neutron already has a callback system [link-to: callbacks.rst] for
in-process resource callbacks where publishers and subscribers are able
to publish, subscribe and extend resources.
This system is different, and is intended to be used for inter-process
callbacks, via the messaging fanout mechanisms.
In Neutron, agents may need to subscribe to specific resource details which
may change over time. And the purpose of this messaging callback system
is to allow agent subscription to those resources without the need to extend
modify existing RPC calls, or creating new RPC messages.
A few resource which can benefit of this system:
* security groups members
* security group rules,
* QoS policies.
Using a remote publisher/subscriber pattern, the information about such
resources could be published using fanout queues to all interested nodes,
minimizing messaging requests from agents to server since the agents
get subscribed for their whole lifecycle (unless they unsubscribe).
Within an agent, there could be multiple subscriber callbacks to the same
resource events, the resources updates would be dispatched to the subscriber
callbacks from a single message. Any update would come in a single message,
doing only a single oslo versioned objects deserialization on each receiving
agent.
This publishing/subscription mechanism is highly dependent on the format
of the resources passed around. This is why the library only allows
versioned objects to be published and subscribed. Oslo versioned objects
allow object version down/up conversion. #[vo_mkcompat]_ #[vo_mkcptests]_
For the VO's versioning schema look here: #[vo_versioning]_
versioned_objects serialization/deserialization with the
obj_to_primitive(target_version=..) and primitive_to_obj() #[ov_serdes]_
methods is used internally to convert/retrieve objects before/after messaging.
Considering rolling upgrades, there are several scenarios to look at:
* publisher (generally neutron-server or a service) and subscriber (agent)
know the same version of the objects, so they serialize, and deserialize
without issues.
* publisher knows (and sends) an older version of the object, subscriber
will get the object updated to latest version on arrival before any
callback is called.
* publisher sends a newer version of the object, subscriber won't be able
to deserialize the object, in this case (PLEASE DISCUSS), we can think of two
strategies:
a) During upgrades, we pin neutron-server to a compatible version for resource
fanout updates, and server sends both the old, and the newer version to
different topic, queues. Old agents receive the updates on the old version
topic, new agents receive updates on the new version topic.
When the whole system upgraded, we un-pin the compatible version fanout.
A variant of this could be using a single fanout queue, and sending the
pinned version of the object to all. Newer agents can deserialize to the
latest version and upgrade any fields internally. Again at the end, we
unpin the version and restart the service.
b) The subscriber will rpc call the publisher to start publishing also a downgraded
version of the object on every update on a separate queue. The complication
of this version, is the need to ignore new version objects as long as we keep
receiving the downgraded ones, and otherwise resend the request to send the
downgraded objects after a certain timeout (thinking of the case where the
request for downgraded queue is done, but the publisher restarted).
This approach is more complicated to implement, but more automated from the
administrator point of view. We may want to look into it as a second step
from a
c) The subscriber will send a registry.get_info for the latest specific version
he knows off. This can have scalability issues during upgrade as any outdated
agent will require a flow of two messages (request, and response). This is
indeed very bad at scale if you have hundreds or thousands of agents.
Option a seems like a reasonable strategy, similar to what nova does now with
versioned objects.
Serialized versioned objects look like::
{'versioned_object.version': '1.0',
'versioned_object.name': 'QoSProfile',
'versioned_object.data': {'rules': [
{'versioned_object.version': '1.0',
'versioned_object.name': 'QoSRule',
'versioned_object.data': {'name': u'a'},
'versioned_object.namespace': 'versionedobjects'}
],
'uuid': u'abcde',
'name': u'aaa'},
'versioned_object.namespace': 'versionedobjects'}
Topic names for the fanout queues
=================================
if we adopted option a:
neutron-<resouce_type>_<resource_id>-<vo_version>
[neutron-<resouce_type>_<resource_id>-<vo_version_compat>]
if we adopted option b for rolling upgrades:
neutron-<resource_type>-<resource_id>
neutron-<resource_type>-<resource_id>-<vo_version>
for option c, just:
neutron-<resource_type>-<resource_id>
Subscribing to resources
========================
Imagine that you have agent A, which just got to handle a new port, which
has an associated security group, and QoS policy.
The agent code processing port updates may look like::
from neutron.rpc_resources import events
from neutron.rpc_resources import resources
from neutron.rpc_resources import registry
def process_resource_updates(resource_type, resource_id, resource_list, action_type):
# send to the right handler which will update any control plane
# details related to the updated resource...
def port_update(...):
# here we extract sg_id and qos_policy_id from port..
registry.subscribe(resources.SG_RULES, sg_id,
callback=process_resource_updates)
sg_rules = registry.get_info(resources.SG_RULES, sg_id)
registry.subscribe(resources.SG_MEMBERS, sg_id,
callback=process_resource_updates)
sg_members = registry.get_info(resources.SG_MEMBERS, sg_id)
registry.subscribe(resources.QOS_RULES, qos_policy_id,
callback=process_resource_updates)
qos_rules = registry.get_info(resources.QOS_RULES, qos_policy_id,
callback=process_resource_updates)
cleanup_subscriptions()
def cleanup_subscriptions()
sg_ids = determine_unreferenced_sg_ids()
qos_policy_id = determine_unreferenced_qos_policy_ids()
registry.unsubscribe_info(resource.SG_RULES, sg_ids)
registry.unsubscribe_info(resource.SG_MEMBERS, sg_ids)
registry.unsubscribe_info(resource.QOS_RULES, qos_policy_id)
Another unsubscription strategy could be to lazily unsubscribe resources when
we receive updates for them, and we discover that they are not needed anymore.
Deleted resources are automatically unsubscribed as we receive the delete event.
NOTE(irenab): this could be extended to core resources like ports, making use
of the standard neutron in-process callbacks at server side and propagating
AFTER_UPDATE events, for example, but we may need to wait until those callbacks
are used with proper versioned objects.
Unsubscribing to resources
==========================
There are a few options to unsubscribe registered callbacks:
* unsubscribe_resource_id(): it selectively unsubscribes an specific
resource type + id.
* unsubscribe_resource_type(): it unsubscribes from an specific resource type,
any ID.
* unsubscribe_all(): it unsubscribes all subscribed resources and ids.
Sending resource updates
========================
On the server side, resource updates could come from anywhere, a service plugin,
an extension, anything that updates the resource and that it's of any interest
to the agents.
The server/publisher side may look like::
from neutron.rpc_resources import events
from neutron.rpc_resources import resources
from neutron.rpc_resources import registry as rpc_registry
def add_qos_x_rule(...):
update_the_db(...)
send_rpc_updates_on_qos_policy(qos_policy_id)
def del_qos_x_rule(...):
update_the_db(...)
send_rpc_deletion_of_qos_policy(qos_policy_id)
def send_rpc_updates_on_qos_policy(qos_policy_id):
rules = get_qos_policy_rules_versioned_object(qos_policy_id)
rpc_registry.notify(resources.QOS_RULES, qos_policy_id, rules, events.UPDATE)
def send_rpc_deletion_of_qos_policy(qos_policy_id):
rpc_registry.notify(resources.QOS_RULES, qos_policy_id, None, events.DELETE)
# This part is added for the registry mechanism, to be able to request
# older versions of the notified objects if any oudated agent requires
# them.
def retrieve_older_version_callback(qos_policy_id, version):
return get_qos_policy_rules_versioned_object(qos_policy_id, version)
rpc_registry.register_retrieve_callback(resource.QOS_RULES,
retrieve_older_version_callback)
References
==========
.. [#ov_serdes] https://github.com/openstack/oslo.versionedobjects/blob/master/oslo_versionedobjects/tests/test_objects.py#L621
.. [#vo_mkcompat] https://github.com/openstack/oslo.versionedobjects/blob/master/oslo_versionedobjects/base.py#L460
.. [#vo_mkcptests] https://github.com/openstack/oslo.versionedobjects/blob/master/oslo_versionedobjects/tests/test_objects.py#L111
.. [#vo_versioning] https://github.com/openstack/oslo.versionedobjects/blob/master/oslo_versionedobjects/base.py#L236

2
etc/neutron.conf

@ -75,7 +75,7 @@
# of its entrypoint name.
#
# service_plugins =
# Example: service_plugins = router,firewall,lbaas,vpnaas,metering
# Example: service_plugins = router,firewall,lbaas,vpnaas,metering,qos
# Paste configuration file
# api_paste_config = api-paste.ini

4
etc/neutron/plugins/ml2/openvswitch_agent.ini

@ -142,6 +142,10 @@
# It should be false when you use nova security group.
# enable_security_group = True
[qos]
# QoS agent driver
# agent_driver = ovs
#-----------------------------------------------------------------------------
# Sample Configurations.
#-----------------------------------------------------------------------------

6
etc/policy.json

@ -39,12 +39,14 @@
"get_network:provider:physical_network": "rule:admin_only",
"get_network:provider:segmentation_id": "rule:admin_only",
"get_network:queue_id": "rule:admin_only",
"get_network:qos_policy_id": "rule:admin_only",
"create_network:shared": "rule:admin_only",
"create_network:router:external": "rule:admin_only",
"create_network:segments": "rule:admin_only",
"create_network:provider:network_type": "rule:admin_only",
"create_network:provider:physical_network": "rule:admin_only",
"create_network:provider:segmentation_id": "rule:admin_only",
"create_network:qos_policy_id": "rule:admin_only",
"update_network": "rule:admin_or_owner",
"update_network:segments": "rule:admin_only",
"update_network:shared": "rule:admin_only",
@ -52,6 +54,7 @@
"update_network:provider:physical_network": "rule:admin_only",
"update_network:provider:segmentation_id": "rule:admin_only",
"update_network:router:external": "rule:admin_only",
"update_network:qos_policy_id": "rule:admin_only",
"delete_network": "rule:admin_or_owner",
"create_port": "",
@ -62,12 +65,14 @@
"create_port:binding:profile": "rule:admin_only",
"create_port:mac_learning_enabled": "rule:admin_or_network_owner or rule:context_is_advsvc",
"create_port:allowed_address_pairs": "rule:admin_or_network_owner",
"create_port:qos_policy_id": "rule:admin_only",
"get_port": "rule:admin_or_owner or rule:context_is_advsvc",
"get_port:queue_id": "rule:admin_only",
"get_port:binding:vif_type": "rule:admin_only",
"get_port:binding:vif_details": "rule:admin_only",
"get_port:binding:host_id": "rule:admin_only",
"get_port:binding:profile": "rule:admin_only",
"get_port:qos_policy_id": "rule:admin_only",
"update_port": "rule:admin_or_owner or rule:context_is_advsvc",
"update_port:mac_address": "rule:admin_only or rule:context_is_advsvc",
"update_port:fixed_ips": "rule:admin_or_network_owner or rule:context_is_advsvc",
@ -76,6 +81,7 @@
"update_port:binding:profile": "rule:admin_only",
"update_port:mac_learning_enabled": "rule:admin_or_network_owner or rule:context_is_advsvc",
"update_port:allowed_address_pairs": "rule:admin_or_network_owner",
"update_port:qos_policy_id": "rule:admin_only",
"delete_port": "rule:admin_or_owner or rule:context_is_advsvc",
"get_router:ha": "rule:admin_only",

75
neutron/agent/common/ovs_lib.py

@ -497,6 +497,81 @@ class OVSBridge(BaseOVS):
txn.add(self.ovsdb.db_set('Controller',
controller_uuid, *attr))
def _create_qos_bw_limit_queue(self, port_name, max_bw_in_bits,
max_burst_in_bits):
external_ids = {'id': port_name}
queue_other_config = {'min-rate': max_bw_in_bits,
'max-rate': max_bw_in_bits,
'burst': max_burst_in_bits}
self.ovsdb.db_create(
'Queue', external_ids=external_ids,
other_config=queue_other_config).execute(check_error=True)
def _create_qos_bw_limit_profile(self, port_name, max_bw_in_bits):
external_ids = {'id': port_name}
queue = self.ovsdb.db_find(
'Queue',
('external_ids', '=', {'id': port_name}),
columns=['_uuid']).execute(
check_error=True)
queues = {}
queues[0] = queue[0]['_uuid']
qos_other_config = {'max-rate': max_bw_in_bits}
self.ovsdb.db_create('QoS', external_ids=external_ids,
other_config=qos_other_config,
type='linux-htb',
queues=queues).execute(check_error=True)
def create_qos_bw_limit_for_port(self, port_name, max_kbps,
max_burst_kbps):
# TODO(QoS) implement this with transactions,
# or roll back on failure
max_bw_in_bits = str(max_kbps * 1000)
max_burst_in_bits = str(max_burst_kbps * 1000)
self._create_qos_bw_limit_queue(port_name, max_bw_in_bits,
max_burst_in_bits)
self._create_qos_bw_limit_profile(port_name, max_bw_in_bits)
qos = self.ovsdb.db_find('QoS',
('external_ids', '=', {'id': port_name}),
columns=['_uuid']).execute(check_error=True)
qos_profile = qos[0]['_uuid']
self.set_db_attribute('Port', port_name, 'qos', qos_profile,
check_error=True)
def get_qos_bw_limit_for_port(self, port_name):
res = self.ovsdb.db_find(
'Queue',
('external_ids', '=', {'id': port_name}),
columns=['other_config']).execute(check_error=True)
if res is None or len(res) == 0:
return None, None
other_config = res[0]['other_config']
max_kbps = int(other_config['max-rate']) / 1000
max_burst_kbps = int(other_config['burst']) / 1000
return max_kbps, max_burst_kbps
def del_qos_bw_limit_for_port(self, port_name):
qos = self.ovsdb.db_find('QoS',
('external_ids', '=', {'id': port_name}),
columns=['_uuid']).execute(check_error=True)
qos_row = qos[0]['_uuid']
queue = self.ovsdb.db_find('Queue',
('external_ids', '=', {'id': port_name}),
columns=['_uuid']).execute(check_error=True)
queue_row = queue[0]['_uuid']
with self.ovsdb.transaction(check_error=True) as txn:
txn.add(self.ovsdb.db_set('Port', port_name, ('qos', [])))
txn.add(self.ovsdb.db_destroy('QoS', qos_row))
txn.add(self.ovsdb.db_destroy('Queue', queue_row))
def __enter__(self):
self.create()
return self

0
neutron/agent/l2/__init__.py Normal file

59
neutron/agent/l2/agent_extension.py Normal file

@ -0,0 +1,59 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
# 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.
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class AgentCoreResourceExtension(object):
"""Define stable abstract interface for Agent extension.
An agent extension extends the agent core functionality.
"""
def initialize(self):
"""Perform agent core resource extension initialization.
Called after all extensions have been loaded.
No abstract methods defined below will be
called prior to this method being called.
"""
pass
def handle_network(self, context, data):
"""handle agent extension for network.
:param context - rpc context
:param data - network data
"""
pass
def handle_subnet(self, context, data):
"""handle agent extension for subnet.
:param context - rpc context
:param data - subnet data
"""
pass
def handle_port(self, context, data):
"""handle agent extension for port.
:param context - rpc context
:param data - port data
"""
pass

73
neutron/agent/l2/agent_extensions_manager.py Normal file

@ -0,0 +1,73 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
# 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
import stevedore
from neutron.i18n import _LE, _LI
LOG = log.getLogger(__name__)
# TODO(QoS) add unit tests to Agent extensions mgr
class AgentExtensionsManager(stevedore.named.NamedExtensionManager):
"""Manage agent extensions."""
def __init__(self):
# Ordered list of agent extensions, defining
# the order in which the agent extensions are called.
#TODO(QoS): get extensions from config
agent_extensions = ('qos', )
LOG.info(_LI("Configured agent extensions names: %s"),
agent_extensions)
super(AgentExtensionsManager, self).__init__(
'neutron.agent.l2.extensions', agent_extensions,
invoke_on_load=True, name_order=True)
LOG.info(_LI("Loaded agent extensions names: %s"), self.names())
def _call_on_agent_extensions(self, method_name, context, data):
"""Helper method for calling a method across all agent extensions."""
for extension in self:
try:
getattr(extension.obj, method_name)(context, data)
# TODO(QoS) add agent extensions exception and catch them here
except AttributeError:
LOG.exception(
_LE("Agent Extension '%(name)s' failed in %(method)s"),
{'name': extension.name, 'method': method_name}
)
def initialize(self):
# Initialize each agent extension in the list.
for extension in self:
LOG.info(_LI("Initializing agent extension '%s'"), extension.name)
extension.obj.initialize()
def handle_network(self, context, data):
"""Notify all agent extensions to handle network."""
self._call_on_agent_extensions("handle_network", context, data)
def handle_subnet(self, context, data):
"""Notify all agent extensions to handle subnet."""
self._call_on_agent_extensions("handle_subnet", context, data)
def handle_port(self, context, data):
"""Notify all agent extensions to handle port."""
self._call_on_agent_extensions("handle_port", context, data)
#TODO(Qos) we are missing how to handle delete. we can pass action
#type in all the handle methods or add handle_delete_resource methods

0
neutron/agent/l2/extensions/__init__.py Normal file

124
neutron/agent/l2/extensions/qos_agent.py Normal file

@ -0,0 +1,124 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
# 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.
import abc
import collections
from oslo_config import cfg
import six
from neutron.agent.l2 import agent_extension
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron import manager
@six.add_metaclass(abc.ABCMeta)
class QosAgentDriver(object):
"""Define stable abstract interface for Qos Agent Driver.
Qos Agent driver defines the interface to be implemented by Agent
for applying Qos Rules on a port.
"""
@abc.abstractmethod
def initialize(self):
"""Perform Qos agent driver initialization.
"""
pass
@abc.abstractmethod
def create(self, port, rules):
"""Apply Qos rules on port for the first time.
:param port: port object.
:param rules: the list of rules to apply on port.
"""
#TODO(Qos) we may want to provide default implementations of calling
#delete and then update
pass
@abc.abstractmethod
def update(self, port, rules):
"""Apply Qos rules on port.
:param port: port object.
:param rules: the list of rules to be apply on port.
"""
pass
@abc.abstractmethod
def delete(self, port, rules):
"""Remove Qos rules from port.
:param port: port object.
:param rules: the list of rules to be removed from port.
"""
pass
class QosAgentExtension(agent_extension.AgentCoreResourceExtension):
def initialize(self):
"""Perform Agent Extension initialization.
"""
super(QosAgentExtension, self).initialize()
self.resource_rpc = resources_rpc.ResourcesServerRpcApi()
self.qos_driver = manager.NeutronManager.load_class_for_provider(
'neutron.qos.agent_drivers', cfg.CONF.qos.agent_driver)()
self.qos_driver.initialize()
self.qos_policy_ports = collections.defaultdict(dict)
self.known_ports = set()
def handle_port(self, context, port):
"""Handle agent qos extension for port.
This method subscribes to qos_policy_id changes
with a callback and get all the qos_policy_ports and apply
them using the qos driver.
Updates and delete event should be handle by the registered
callback.
"""
port_id = port['port_id']
qos_policy_id = port.get('qos_policy_id')
if qos_policy_id is None:
#TODO(QoS): we should also handle removing policy
return
#Note(moshele) check if we have seen this port
#and it has the same policy we do nothing.
if (port_id in self.known_ports and
port_id in self.qos_policy_ports[qos_policy_id]):
return
self.qos_policy_ports[qos_policy_id][port_id] = port
self.known_ports.add(port_id)
#TODO(QoS): handle updates when implemented
# we have two options:
# 1. to add new api for subscribe
# registry.subscribe(self._process_rules_updates,
# resources.QOS_RULES, qos_policy_id)
# 2. combine get_info rpc to also subscribe to the resource
qos_rules = self.resource_rpc.get_info(
context, resources.QOS_POLICY, qos_policy_id)
self._process_rules_updates(
port, resources.QOS_POLICY, qos_policy_id,
qos_rules, 'create')
def _process_rules_updates(
self, port, resource_type, resource_id,
qos_rules, action_type):
getattr(self.qos_driver, action_type)(port, qos_rules)

23
neutron/agent/ovsdb/api.py

@ -161,6 +161,29 @@ class API(object):
:returns: :class:`Command` with field value result
"""
@abc.abstractmethod
def db_create(self, table, **col_values):
"""Create a command to create new record
:param table: The OVS table containing the record to be created
:type table: string
:param col_values: The columns and their associated values
to be set after create
:type col_values: Dictionary of columns id's and values
:returns: :class:`Command` with no result
"""
@abc.abstractmethod
def db_destroy(self, table, record):
"""Create a command to destroy a record
:param table: The OVS table containing the record to be destroyed
:type table: string
:param record: The record id (name/uuid) to be destroyed
:type record: uuid/string
:returns: :class:`Command` with no result
"""
@abc.abstractmethod
def db_set(self, table, record, *col_values):
"""Create a command to set fields in a record

6
neutron/agent/ovsdb/impl_idl.py

@ -168,6 +168,12 @@ class OvsdbIdl(api.API):
def br_set_external_id(self, name, field, value):
return cmd.BrSetExternalIdCommand(self, name, field, value)
def db_create(self, table, **col_values):
return cmd.DbCreateCommand(self, table, **col_values)
def db_destroy(self, table, record):
return cmd.DbDestroyCommand(self, table, record)
def db_set(self, table, record, *col_values):
return cmd.DbSetCommand(self, table, record, *col_values)

16
neutron/agent/ovsdb/impl_vsctl.py

@ -184,6 +184,15 @@ class OvsdbVsctl(ovsdb.API):
return BaseCommand(self.context, 'br-get-external-id',
args=[name, field])
def db_create(self, table, **col_values):
args = [table]
args += _set_colval_args(*col_values.items())
return BaseCommand(self.context, 'create', args=args)
def db_destroy(self, table, record):
args = [table, record]
return BaseCommand(self.context, 'destroy', args=args)
def db_set(self, table, record, *col_values):
args = [table, record]
args += _set_colval_args(*col_values)
@ -259,8 +268,11 @@ def _set_colval_args(*col_values):
col, k, op, ovsdb.py_to_val(v)) for k, v in val.items()]
elif (isinstance(val, collections.Sequence)
and not isinstance(val, six.string_types)):
args.append(
"%s%s%s" % (col, op, ",".join(map(ovsdb.py_to_val, val))))
if len(val) == 0:
args.append("%s%s%s" % (col, op, "[]"))
else:
args.append(
"%s%s%s" % (col, op, ",".join(map(ovsdb.py_to_val, val))))
else:
args.append("%s%s%s" % (col, op, ovsdb.py_to_val(val)))
return args

24
neutron/agent/ovsdb/native/commands.py

@ -148,6 +148,30 @@ class BrSetExternalIdCommand(BaseCommand):
br.external_ids = external_ids
class DbCreateCommand(BaseCommand):
def __init__(self, api, table, **columns):
super(DbCreateCommand, self).__init__(api)
self.table = table
self.columns = columns
def run_idl(self, txn):
row = txn.insert(self.api._tables[self.table])
for col, val in self.columns.items():
setattr(row, col, val)
self.result = row
class DbDestroyCommand(BaseCommand):
def __init__(self, api, table, record):
super(DbDestroyCommand, self).__init__(api)
self.table = table
self.record = record
def run_idl(self, txn):
record = idlutils.row_by_record(self.api.idl, self.table, self.record)
record.delete()
class DbSetCommand(BaseCommand):
def __init__(self, api, table, record, *col_values):
super(DbSetCommand, self).__init__(api)

0
neutron/api/rpc/callbacks/__init__.py Normal file

19
neutron/api/rpc/callbacks/events.py Normal file

@ -0,0 +1,19 @@
# 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.
UPDATED = 'updated'
DELETED = 'deleted'
VALID = (
UPDATED,
DELETED
)

68
neutron/api/rpc/callbacks/registry.py Normal file

@ -0,0 +1,68 @@
# 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.api.rpc.callbacks import resource_manager
# TODO(ajo): consider adding locking
CALLBACK_MANAGER = None
def _get_resources_callback_manager():
global CALLBACK_MANAGER
if CALLBACK_MANAGER is None:
CALLBACK_MANAGER = resource_manager.ResourcesCallbacksManager()
return CALLBACK_MANAGER
#resource implementation callback registration functions
def get_info(resource_type, resource_id, **kwargs):
"""Get information about resource type with resource id.
The function will check the providers for an specific remotable
resource and get the resource.
:returns: an oslo versioned object.
"""
callback = _get_resources_callback_manager().get_callback(resource_type)
if callback:
return callback(resource_type, resource_id, **kwargs)
def register_provider(callback, resource_type):
_get_resources_callback_manager().register(callback, resource_type)
# resource RPC callback for pub/sub
#Agent side
def subscribe(callback, resource_type, resource_id):
#TODO(QoS): we have to finish the real update notifications
raise NotImplementedError("we should finish update notifications")
def unsubscribe(callback, resource_type, resource_id):
#TODO(QoS): we have to finish the real update notifications
raise NotImplementedError("we should finish update notifications")
def unsubscribe_all():
#TODO(QoS): we have to finish the real update notifications
raise NotImplementedError("we should finish update notifications")
#Server side
def notify(resource_type, event, obj):
#TODO(QoS): we have to finish the real update notifications
raise NotImplementedError("we should finish update notifications")
def clear():
_get_resources_callback_manager().clear()

69
neutron/api/rpc/callbacks/resource_manager.py Normal file

@ -0,0 +1,69 @@
# 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 collections
from oslo_log import log as logging
from neutron.api.rpc.callbacks import resources
from neutron.callbacks import exceptions
LOG = logging.getLogger(__name__)
class ResourcesCallbacksManager(object):
"""A callback system that allows information providers in a loose manner.
"""
def __init__(self):
self.clear()
def register(self, callback, resource):
"""register callback for a resource .
One callback can be register to a resource
:param callback: the callback. It must raise or return a dict.
:param resource: the resource. It must be a valid resource.
"""
LOG.debug("register: %(callback)s %(resource)s",
{'callback': callback, 'resource': resource})
if resource not in resources.VALID:
raise exceptions.Invalid(element='resource', value=resource)
self._callbacks[resource] = callback
def unregister(self, resource):
"""Unregister callback from the registry.
:param callback: the callback.
:param resource: the resource.
"""
LOG.debug("Unregister: %(resource)s",
{'resource': resource})
if resource not in resources.VALID:
raise exceptions.Invalid(element='resource', value=resource)
self._callbacks[resource] = None
def clear(self):
"""Brings the manager to a clean slate."""
self._callbacks = collections.defaultdict(dict)
def get_callback(self, resource):
"""Return the callback if found, None otherwise.
:param resource: the resource. It must be a valid resource.
"""
if resource not in resources.VALID:
raise exceptions.Invalid(element='resource', value=resource)
return self._callbacks[resource]

19
neutron/api/rpc/callbacks/resources.py Normal file

@ -0,0 +1,19 @@
# 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.
QOS_POLICY = 'qos-policy'
QOS_RULE = 'qos-rule'
VALID = (
QOS_POLICY,
QOS_RULE,
)

71
neutron/api/rpc/handlers/resources_rpc.py Executable file

@ -0,0 +1,71 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
# 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 helpers as log_helpers
from oslo_log import log as logging
import oslo_messaging
from neutron.api.rpc.callbacks import registry
from neutron.common import constants
from neutron.common import rpc as n_rpc
from neutron.common import topics
LOG = logging.getLogger(__name__)
class ResourcesServerRpcApi(object):
"""Agent-side RPC (stub) for agent-to-plugin interaction.
This class implements the client side of an rpc interface. The server side
can be found below: ResourcesServerRpcCallback. For more information on
changing rpc interfaces, see doc/source/devref/rpc_api.rst.
"""
def __init__(self):
target = oslo_messaging.Target(
topic=topics.PLUGIN, version='1.0',
namespace=constants.RPC_NAMESPACE_RESOURCES)
self.client = n_rpc.get_client(target)
@log_helpers.log_method_call
def get_info(self, context, resource_type, resource_id):
cctxt = self.client.prepare()
#TODO(Qos): add deserialize version object
return cctxt.call(context, 'get_info',
resource_type=resource_type, resource_id=resource_id)
class ResourcesServerRpcCallback(object):
"""Plugin-side RPC (implementation) for agent-to-plugin interaction.
This class implements the server side of an rpc interface. The client side
can be found above: ResourcesServerRpcApi. For more information on
changing rpc interfaces, see doc/source/devref/rpc_api.rst.
"""
# History
# 1.0 Initial version
target = oslo_messaging.Target(
version='1.0', namespace=constants.RPC_NAMESPACE_RESOURCES)
def get_info(self, context, resource_type, resource_id):
kwargs = {'context': context}
#TODO(Qos): add serialize version object
return registry.get_info(
resource_type,
resource_id,
**kwargs)

2
neutron/callbacks/resources.py

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
NETWORK = 'network'
PORT = 'port'
ROUTER = 'router'
ROUTER_GATEWAY = 'router_gateway'
@ -19,6 +20,7 @@ SECURITY_GROUP_RULE = 'security_group_rule'
SUBNET = 'subnet'
VALID = (
NETWORK,
PORT,
ROUTER,
ROUTER_GATEWAY,

2
neutron/common/constants.py

@ -175,6 +175,8 @@ RPC_NAMESPACE_SECGROUP = None
RPC_NAMESPACE_DVR = None
# RPC interface for reporting state back to the plugin
RPC_NAMESPACE_STATE = None
# RPC interface for agent to plugin resources API
RPC_NAMESPACE_RESOURCES = None
# Default network MTU value when not configured
DEFAULT_NETWORK_MTU = 0

4
neutron/common/exceptions.py

@ -470,3 +470,7 @@ class DeviceNotFoundError(NeutronException):
class NetworkSubnetPoolAffinityError(BadRequest):
message = _("Subnets hosted on the same network must be allocated from "
"the same subnet pool")
class ObjectActionError(NeutronException):
message = _('Object action %(action)s failed because: %(reason)s')

1
neutron/common/topics.py

@ -19,6 +19,7 @@ PORT = 'port'
SECURITY_GROUP = 'security_group'
L2POPULATION = 'l2population'
DVR = 'dvr'
RESOURCES = 'resources'
CREATE = 'create'
DELETE = 'delete'

4
neutron/common/utils.py

@ -434,3 +434,7 @@ class DelayedStringRenderer(object):
def __str__(self):
return str(self.function(*self.args, **self.kwargs))
def camelize(s):
return ''.join(s.replace('_', ' ').title().split())

41
neutron/db/api.py

@ -20,9 +20,12 @@ from oslo_config import cfg
from oslo_db import api as oslo_db_api
from oslo_db import exception as os_db_exception
from oslo_db.sqlalchemy import session
from oslo_utils import uuidutils
from sqlalchemy import exc
from sqlalchemy import orm
from neutron.db import common_db_mixin
_FACADE = None
@ -88,3 +91,41 @@ class convert_db_exception_to_retry(object):
except self.to_catch as e:
raise os_db_exception.RetryRequest(e)
return wrapper
# Common database operation implementations
def get_object(context, model, **kwargs):
with context.session.begin(subtransactions=True):
return (common_db_mixin.model_query(context, model)
.filter_by(**kwargs)
.first())
def get_objects(context, model, **kwargs):
with context.session.begin(subtransactions=True):
return (common_db_mixin.model_query(context, model)
.filter_by(**kwargs)
.all())
def create_object(context, model, values):
with context.session.begin(subtransactions=True):
if 'id' not in values:
values['id'] = uuidutils.generate_uuid()
db_obj = model(**values)
context.session.add(db_obj)
return db_obj.__dict__
def update_object(context, model, id, values):
with context.session.begin(subtransactions=True):
db_obj = get_object(context, model, id=id)
db_obj.update(values)
db_obj.save(session=context.session)
return db_obj.__dict__
def delete_object(context, model, id):
with context.session.begin(subtransactions=True):
db_obj = get_object(context, model, id=id)
context.session.delete(db_obj)

2
neutron/db/migration/alembic_migrations/versions/HEADS

@ -1,3 +1,3 @@
2a16083502f3
8675309a5c4f
48153cb5f051
kilo

77
neutron/db/migration/alembic_migrations/versions/liberty/expand/48153cb5f051_qos_db_changes.py Executable file

@ -0,0 +1,77 @@
# Copyright 2015 Huawei Technologies India Pvt Ltd, Inc
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""qos db changes
Revision ID: 48153cb5f051
Revises: 8675309a5c4f
Create Date: 2015-06-24 17:03:34.965101
"""
# revision identifiers, used by Alembic.
revision = '48153cb5f051'
down_revision = '8675309a5c4f'
from alembic import op
import sqlalchemy as sa
from neutron.api.v2 import attributes as attrs
def upgrade():
op.create_table(
'qos_policies',
sa.Column('id', sa.String(length=36), primary_key=True),
sa.Column('name', sa.String(length=attrs.NAME_MAX_LEN)),
sa.Column('description', sa.String(length=attrs.DESCRIPTION_MAX_LEN)),
sa.Column('shared', sa.Boolean()),
sa.Column('tenant_id', sa.String(length=attrs.TENANT_ID_MAX_LEN),
index=True))
op.create_table(
'qos_network_policy_bindings',
sa.Column('policy_id', sa.String(length=36),
sa.ForeignKey('qos_policies.id', ondelete='CASCADE'),
nullable=False),
sa.Column('network_id', sa.String(length=36),
sa.ForeignKey('networks.id', ondelete='CASCADE'),
nullable=False, unique=True))
op.create_table(
'qos_port_policy_bindings',
sa.Column('policy_id', sa.String(length=36),
sa.ForeignKey('qos_policies.id', ondelete='CASCADE'),
nullable=False),
sa.Column('port_id', sa.String(length=36),
sa.ForeignKey('ports.id', ondelete='CASCADE'),
nullable=False, unique=True))
op.create_table(
'qos_rules',
sa.Column('id', sa.String(length=36), primary_key=True),
sa.Column('qos_policy_id', sa.String(length=36),
sa.ForeignKey('qos_policies.id', ondelete='CASCADE'),
nullable=False),
sa.Column('type', sa.String(length=255)))
op.create_table(
'qos_bandwidth_limit_rules',
sa.Column('id', sa.String(length=36),
sa.ForeignKey('qos_rules.id', ondelete='CASCADE'),
nullable=False,
primary_key=True),
sa.Column('max_kbps', sa.Integer()),
sa.Column('max_burst_kbps', sa.Integer()))

1
neutron/db/migration/models/head.py

@ -41,6 +41,7 @@ from neutron.db import model_base
from neutron.db import models_v2 # noqa
from neutron.db import portbindings_db # noqa
from neutron.db import portsecurity_db # noqa
from neutron.db.qos import models as qos_models # noqa
from neutron.db import quota_db # noqa
from neutron.db import rbac_db_models # noqa
from neutron.db import securitygroups_db # noqa

0
neutron/db/qos/__init__.py Normal file

44
neutron/db/qos/api.py Normal file

@ -0,0 +1,44 @@
# 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 common_db_mixin as db
from neutron.db.qos import models
def create_policy_network_binding(context, policy_id, network_id):
with context.session.begin(subtransactions=True):
db_obj = models.QosNetworkPolicyBinding(policy_id=policy_id,
network_id=network_id)
context.session.add(db_obj)
def delete_policy_network_binding(context, policy_id, network_id):
with context.session.begin(subtransactions=True):
db_object = (db.model_query(context, models.QosNetworkPolicyBinding)
.filter_by(policy_id=policy_id,
network_id=network_id).one())
context.session.delete(db_object)
def create_policy_port_binding(context, policy_id, port_id):
with context.session.begin(subtransactions=True):
db_obj = models.QosPortPolicyBinding(policy_id=policy_id,
port_id=port_id)
context.session.add(db_obj)
def delete_policy_port_binding(context, policy_id, port_id):
with context.session.begin(subtransactions=True):
db_object = (db.model_query(context, models.QosPortPolicyBinding)
.filter_by(policy_id=policy_id,
port_id=port_id).one())
context.session.delete(db_object)

89
neutron/db/qos/models.py Executable file

@ -0,0 +1,89 @@
# Copyright 2015 Huawei Technologies India Pvt Ltd, 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
import sqlalchemy as sa
from neutron.api.v2 import attributes as attrs
from neutron.db import model_base
from neutron.db import models_v2
LOG = logging.getLogger(__name__)
class QosPolicy(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
__tablename__ = 'qos_policies'
name = sa.Column(sa.String(attrs.NAME_MAX_LEN))
description = sa.Column(sa.String(attrs.DESCRIPTION_MAX_LEN))
shared = sa.Column(sa.Boolean)
class QosNetworkPolicyBinding(model_base.BASEV2):
__tablename__ = 'qos_network_policy_bindings'
policy_id = sa.Column(sa.String(36),
sa.ForeignKey('qos_policies.id',
ondelete='CASCADE'),
nullable=False,
primary_key=True)
network_id = sa.Column(sa.String(36),
sa.ForeignKey('networks.id',
ondelete='CASCADE'),
nullable=False,
unique=True,
primary_key=True)
network = sa.orm.relationship(
models_v2.Network,
backref=sa.orm.backref("qos_policy_binding", uselist=False,
cascade='delete', lazy='joined'))
class QosPortPolicyBinding(model_base.BASEV2):
__tablename__ = 'qos_port_policy_bindings'
policy_id = sa.Column(sa.String(36),
sa.ForeignKey('qos_policies.id',
ondelete='CASCADE'),
nullable=False,
primary_key=True)
port_id = sa.Column(sa.String(36),
sa.ForeignKey('ports.id',
ondelete='CASCADE'),
nullable=False,
unique=True,
primary_key=True)
port = sa.orm.relationship(
models_v2.Port,
backref=sa.orm.backref("qos_policy_binding", uselist=False,
cascade='delete', lazy='joined'))
class QosRule(model_base.BASEV2, models_v2.HasId):
__tablename__ = 'qos_rules'
type = sa.Column(sa.String(255))
qos_policy_id = sa.Column(sa.String(36),
sa.ForeignKey('qos_policies.id',
ondelete='CASCADE'),
nullable=False)
class QosBandwidthLimitRule(model_base.BASEV2):
__tablename__ = 'qos_bandwidth_limit_rules'
max_kbps = sa.Column(sa.Integer)
max_burst_kbps = sa.Column(sa.Integer)
id = sa.Column(sa.String(36),
sa.ForeignKey('qos_rules.id',
ondelete='CASCADE'),
nullable=False,
primary_key=True)

240
neutron/extensions/qos.py Normal file

@ -0,0 +1,240 @@
# Copyright (c) 2015 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.
import abc
import itertools
import six
from neutron.api import extensions
from neutron.api.v2 import attributes as attr
from neutron.api.v2 import base
from neutron.api.v2 import resource_helper
from neutron import manager
from neutron.plugins.common import constants
from neutron.services import service_base
QOS_PREFIX = "/qos"
RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth_limit'
VALID_RULE_TYPES = [RULE_TYPE_BANDWIDTH_LIMIT]
# Attribute Map
QOS_RULE_COMMON_FIELDS = {
'id': {'allow_post': False, 'allow_put': False,
'validate': {'type:uuid': None},
'is_visible': True,
'primary_key': True},
'type': {'allow_post': True, 'allow_put': True, 'is_visible': True,
'default': '',
'validate': {'type:values': VALID_RULE_TYPES}},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'is_visible': True},
}
RESOURCE_ATTRIBUTE_MAP = {
'policies': {
'id': {'allow_post': False, 'allow_put': False,
'validate': {'type:uuid': None},
'is_visible': True, 'primary_key': True},
'name': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': '',
'validate': {'type:string': None}},
'description': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': '',
'validate': {'type:string': None}},
'shared': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': False,
'convert_to': attr.convert_to_boolean},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'is_visible': True}
},
'rule_types': {
'type': {'allow_post': False, 'allow_put': False,
'is_visible': True}
}
}
SUB_RESOURCE_ATTRIBUTE_MAP = {
'bandwidth_limit_rules': {
'parent': {'collection_name': 'policies',
'member_name': 'policy'},
'parameters': dict(QOS_RULE_COMMON_FIELDS,
**{'max_kbps': {
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': None,
'validate': {'type:non_negative': None}},
'max_burst_kbps': {
'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': 0,
'validate': {'type:non_negative': None}}})
}
}
QOS_POLICY_ID = "qos_policy_id"
EXTENDED_ATTRIBUTES_2_0 = {
'ports': {QOS_POLICY_ID: {'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'validate': {'type:uuid_or_none': None}}},
'networks': {QOS_POLICY_ID: {'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'validate': {'type:uuid_or_none': None}}}}
class Qos(extensions.ExtensionDescriptor):
"""Quality of service API extension."""
@classmethod
def get_name(cls):
return "qos"
@classmethod
def get_alias(cls):
return "qos"
@classmethod
def get_description(cls):
return "The Quality of Service extension."
@classmethod
def get_updated(cls):
return "2015-06-08T10:00:00-00:00"
@classmethod
def get_plugin_interface(cls):
return QoSPluginBase
@classmethod
def get_resources(cls):
"""Returns Ext Resources."""
special_mappings = {'policies': 'policy'}
plural_mappings = resource_helper.build_plural_mappings(
special_mappings, itertools.chain(RESOURCE_ATTRIBUTE_MAP,
SUB_RESOURCE_ATTRIBUTE_MAP))
attr.PLURALS.update(plural_mappings)
resources = resource_helper.build_resource_info(
plural_mappings,
RESOURCE_ATTRIBUTE_MAP,
constants.QOS,
translate_name=True,
allow_bulk=True)
plugin = manager.NeutronManager.get_service_plugins()[constants.QOS]
for collection_name in SUB_RESOURCE_ATTRIBUTE_MAP:
resource_name = collection_name[:-1]
parent = SUB_RESOURCE_ATTRIBUTE_MAP[collection_name].get('parent')
params = SUB_RESOURCE_ATTRIBUTE_MAP[collection_name].get(
'parameters')
controller = base.create_resource(collection_name, resource_name,
plugin, params,
allow_bulk=True,
parent=parent,
allow_pagination=True,
allow_sorting=True)
resource = extensions.ResourceExtension(
collection_name,
controller, parent,
path_prefix=QOS_PREFIX,
attr_map=params)
resources.append(resource)
return resources
def update_attributes_map(self, attributes, extension_attrs_map=None):
super(Qos, self).update_attributes_map(
attributes, extension_attrs_map=RESOURCE_ATTRIBUTE_MAP)
def get_extended_resources(self, version):
if version == "2.0":
return dict(EXTENDED_ATTRIBUTES_2_0.items() +
RESOURCE_ATTRIBUTE_MAP.items())
else:
return {}
@six.add_metaclass(abc.ABCMeta)
class QoSPluginBase(service_base.ServicePluginBase):
path_prefix = QOS_PREFIX
def get_plugin_description(self):
return "QoS Service Plugin for ports and networks"
def get_plugin_type(self):
return constants.QOS
@abc.abstractmethod
def get_policy(self, context, policy_id, fields=None):
pass
@abc.abstractmethod
def get_policies(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
pass
@abc.abstractmethod
def create_policy(self, context, policy):
pass
@abc.abstractmethod
def update_policy(self, context, policy_id, policy):
pass
@abc.abstractmethod
def delete_policy(self, context, policy_id):
pass
@abc.abstractmethod
def get_policy_bandwidth_limit_rule(self, context, rule_id,
policy_id, fields=None):
pass
@abc.abstractmethod
def get_policy_bandwidth_limit_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
pass
@abc.abstractmethod
def create_policy_bandwidth_limit_rule(self, context, policy_id,
bandwidth_limit_rule):
pass
@abc.abstractmethod
def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id,
bandwidth_limit_rule):
pass
@abc.abstractmethod
def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id):
pass
@abc.abstractmethod
def get_rule_types(self, context, filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
pass

0
neutron/objects/__init__.py Normal file

85
neutron/objects/base.py Normal file

@ -0,0 +1,85 @@
# 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
from oslo_versionedobjects import base as obj_base
import six
from neutron.db import api as db_api
# TODO(QoS): revisit dict compatibility and how we can isolate dict behavior
@six.add_metaclass(abc.ABCMeta)
class NeutronObject(obj_base.VersionedObject,
obj_base.VersionedObjectDictCompat,
obj_base.ComparableVersionedObject):
# should be overridden for all persistent objects
db_model = None
# fields that are not allowed to update
fields_no_update = []
synthetic_fields = []
def from_db_object(self, *objs):
for field in self.fields:
for db_obj in objs:
if field in db_obj:
setattr(self, field, db_obj[field])
break
self.obj_reset_changes()
# TODO(QoS): this should be revisited on how we plan to work with dicts
def to_dict(self):
return dict(self.items())
@classmethod
def get_by_id(cls, context, id):
db_obj = db_api.get_object(context, cls.db_model, id=id)
if db_obj:
obj = cls(context, **db_obj)
obj.obj_reset_changes()
return obj
@classmethod
def get_objects(cls, context, **kwargs):
db_objs = db_api.get_objects(context, cls.db_model, **kwargs)
objs = [cls(context, **db_obj) for db_obj in db_objs]
for obj in objs:
obj.obj_reset_changes()
return objs
def _get_changed_persistent_fields(self):
fields = self.obj_get_changes()
for field in self.synthetic_fields:
if field in fields:
del fields[field]
return fields
def create(self):
fields = self._get_changed_persistent_fields()
db_obj = db_api.create_object(self._context, self.db_model, fields)
self.from_db_object(db_obj)
def update(self):
updates = self._get_changed_persistent_fields()
if updates:
db_obj = db_api.update_object(self._context, self.db_model,
self.id, updates)
self.from_db_object(self, db_obj)
def delete(self):
db_api.delete_object(self._context, self.db_model, self.id)

0
neutron/objects/qos/__init__.py Normal file

113
neutron/objects/qos/policy.py Normal file

@ -0,0 +1,113 @@
# Copyright 2015 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.
import abc
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
import six
from neutron.common import exceptions
from neutron.common import utils
from neutron.db import api as db_api
from neutron.db.qos import api as qos_db_api
from neutron.db.qos import models as qos_db_model
from neutron.extensions import qos as qos_extension
from neutron.objects import base
from neutron.objects.qos import rule as rule_obj_impl
class QosRulesExtenderMeta(abc.ABCMeta):
def __new__(mcs, name, bases, dct):
cls = super(QosRulesExtenderMeta, mcs).__new__(mcs, name, bases, dct)
cls.rule_fields = {}
for rule in qos_extension.VALID_RULE_TYPES:
rule_cls_name = 'Qos%sRule' % utils.camelize(rule)
field = '%s_rules' % rule
cls.fields[field] = obj_fields.ListOfObjectsField(rule_cls_name)
cls.rule_fields[field] = rule_cls_name
cls.synthetic_fields = list(cls.rule_fields.keys())
return cls
@obj_base.VersionedObjectRegistry.register
@six.add_metaclass(QosRulesExtenderMeta)
class QosPolicy(base.NeutronObject):
db_model = qos_db_model.QosPolicy
port_binding_model = qos_db_model.QosPortPolicyBinding
network_binding_model = qos_db_model.QosNetworkPolicyBinding
fields = {
'id': obj_fields.UUIDField(),
'tenant_id': obj_fields.UUIDField(),
'name': obj_fields.StringField(),
'description': obj_fields.StringField(),
'shared': obj_fields.BooleanField()
}
fields_no_update = ['id', 'tenant_id']
def obj_load_attr(self, attrname):
if attrname not in self.rule_fields:
raise exceptions.ObjectActionError(
action='obj_load_attr', reason='unable to load %s' % attrname)
rule_cls = getattr(rule_obj_impl, self.rule_fields[attrname])
rules = rule_cls.get_rules_by_policy(self._context, self.id)
setattr(self, attrname, rules)
self.obj_reset_changes([attrname])
@classmethod
def _get_object_policy(cls, context, model, **kwargs):
binding_db_obj = db_api.get_object(context, model, **kwargs)
# TODO(QoS): rethink handling missing binding case
if binding_db_obj:
return cls.get_by_id(context, binding_db_obj['policy_id'])
@classmethod
def get_network_policy(cls, context, network_id):
return cls._get_object_policy(context, cls.network_binding_model,
network_id=network_id)
@classmethod
def get_port_policy(cls, context, port_id):
return cls._get_object_policy(context, cls.port_binding_model,
port_id=port_id)
def attach_network(self, network_id):
qos_db_api.create_policy_network_binding(self._context,
policy_id=self.id,
network_id=network_id)
def attach_port(self, port_id):
qos_db_api.create_policy_port_binding(self._context,
policy_id=self.id,
port_id=port_id)
def detach_network(self, network_id):
qos_db_api.delete_policy_network_binding(self._context,
policy_id=self.id,
network_id=network_id)
def detach_port(self, port_id):
qos_db_api.delete_policy_port_binding(self._context,
policy_id=self.id,
port_id=port_id)

157
neutron/objects/qos/rule.py Normal file

@ -0,0 +1,157 @@
# Copyright 2015 Huawei Technologies India Pvt Ltd, 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.
import abc
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
import six
from neutron.db import api as db_api
from neutron.db.qos import models as qos_db_model
from neutron.extensions import qos as qos_extension
from neutron.objects import base
@six.add_metaclass(abc.ABCMeta)
class QosRule(base.NeutronObject):
base_db_model = qos_db_model.QosRule
fields = {
'id': obj_fields.UUIDField(),
'type': obj_fields.StringField(),
'qos_policy_id': obj_fields.UUIDField()
}
fields_no_update = ['id', 'tenant_id', 'qos_policy_id']
# each rule subclass should redefine it
rule_type = None
_core_fields = list(fields.keys())
_common_fields = ['id']
@classmethod
def _is_common_field(cls, field):
return field in cls._common_fields
@classmethod
def _is_core_field(cls, field):
return field in cls._core_fields
@classmethod
def _is_addn_field(cls, field):
return not cls._is_core_field(field) or cls._is_common_field(field)
@staticmethod
def _filter_fields(fields, func):
return {
key: val for key, val in fields.items()
if func(key)
}
def _get_changed_core_fields(self):
fields = self.obj_get_changes()
return self._filter_fields(fields, self._is_core_field)
def _get_changed_addn_fields(self):
fields = self.obj_get_changes()
return self._filter_fields(fields, self._is_addn_field)
def _copy_common_fields(self, from_, to_):
for field in self._common_fields:
to_[field] = from_[field]
@classmethod
def get_objects(cls, context, **kwargs):
# TODO(QoS): support searching for subtype fields
db_objs = db_api.get_objects(context, cls.base_db_model, **kwargs)
return [cls.get_by_id(context, db_obj['id']) for db_obj in db_objs]
@classmethod
def get_by_id(cls, context, id):
obj = super(QosRule, cls).get_by_id(context, id)
if obj:
# the object above does not contain fields from base QosRule yet,
# so fetch it and mix its fields into the object
base_db_obj = db_api.get_object(context, cls.base_db_model, id=id)
for field in cls._core_fields:
setattr(obj, field, base_db_obj[field])
obj.obj_reset_changes()
return obj
# TODO(QoS): create and update are not transactional safe
def create(self):
# TODO(QoS): enforce that type field value is bound to specific class
self.type = self.rule_type
# create base qos_rule
core_fields = self._get_changed_core_fields()
base_db_obj = db_api.create_object(
self._context, self.base_db_model, core_fields)
# create type specific qos_..._rule
addn_fields = self._get_changed_addn_fields()
self._copy_common_fields(core_fields, addn_fields)
addn_db_obj = db_api.create_object(
self._context, self.db_model, addn_fields)
# merge two db objects into single neutron one
self.from_db_object(base_db_obj, addn_db_obj)
def update(self):
updated_db_objs = []
# TODO(QoS): enforce that type field cannot be changed
# update base qos_rule, if needed
core_fields = self._get_changed_core_fields()
if core_fields:
base_db_obj = db_api.update_object(
self._context, self.base_db_model, self.id, core_fields)
updated_db_objs.append(base_db_obj)
addn_fields = self._get_changed_addn_fields()
if addn_fields:
addn_db_obj = db_api.update_object(
self._context, self.db_model, self.id, addn_fields)
updated_db_objs.append(addn_db_obj)
# update neutron object with values from both database objects
self.from_db_object(*updated_db_objs)
# delete is the same, additional rule object cleanup is done thru cascading
@classmethod
def get_rules_by_policy(cls, context, policy_id):
return cls.get_objects(context, qos_policy_id=policy_id)
@obj_base.VersionedObjectRegistry.register
class QosBandwidthLimitRule(QosRule):
db_model = qos_db_model.QosBandwidthLimitRule
rule_type = qos_extension.RULE_TYPE_BANDWIDTH_LIMIT
fields = {
'max_kbps': obj_fields.IntegerField(nullable=True),
'max_burst_kbps': obj_fields.IntegerField(nullable=True)
}

4
neutron/plugins/common/constants.py

@ -23,6 +23,7 @@ VPN = "VPN"
METERING = "METERING"
L3_ROUTER_NAT = "L3_ROUTER_NAT"
FLAVORS = "FLAVORS"
QOS = "QOS"
# Maps extension alias to service type
EXT_TO_SERVICE_MAPPING = {
@ -33,7 +34,8 @@ EXT_TO_SERVICE_MAPPING = {
'vpnaas': VPN,
'metering': METERING,
'router': L3_ROUTER_NAT,
'flavors': FLAVORS
'flavors': FLAVORS,
'qos': QOS,
}
# Service operation status constants

6
neutron/plugins/ml2/driver_api.py

@ -911,12 +911,14 @@ class ExtensionDriver(object):
"""
pass
@abc.abstractproperty
@property
def extension_alias(self):
"""Supported extension alias.
Return the alias identifying the core API extension supported
by this driver.
by this driver. Do not declare if API extension handling will
be left to a service plugin, and we just need to provide
core resource extension and updates.
"""
pass

5
neutron/plugins/ml2/drivers/openvswitch/agent/common/config.py

@ -100,7 +100,12 @@ agent_opts = [
"timeout won't be changed"))
]
qos_opts = [
cfg.StrOpt('agent_driver', default='ovs', help=_('QoS agent driver.')),
]
cfg.CONF.register_opts(ovs_opts, "OVS")
cfg.CONF.register_opts(agent_opts, "AGENT")
cfg.CONF.register_opts(qos_opts, "qos")
config.register_agent_state_opts_helper(cfg.CONF)

0
neutron/plugins/ml2/drivers/openvswitch/agent/extension_drivers/__init__.py Normal file

79
neutron/plugins/ml2/drivers/openvswitch/agent/extension_drivers/qos_driver.py Normal file

@ -0,0 +1,79 @@
# Copyright (c) 2015 Openstack Foundation
#
# 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 oslo_log import log as logging
from neutron.agent.common import ovs_lib
from neutron.agent.l2.extensions import qos_agent
from neutron.extensions import qos
LOG = logging.getLogger(__name__)
class QosOVSAgentDriver(qos_agent.QosAgentDriver):
def __init__(self):
super(QosOVSAgentDriver, self).__init__()
# TODO(QoS) check if we can get this configuration
# as constructor arguments
self.br_int_name = cfg.CONF.OVS.integration_bridge
self.br_int = None
self.handlers = {}
def initialize(self):
self.handlers[('update', qos.RULE_TYPE_BANDWIDTH_LIMIT)] = (
self._update_bw_limit_rule)
self.handlers[('create', qos.RULE_TYPE_BANDWIDTH_LIMIT)] = (
self._update_bw_limit_rule)
self.handlers[('delete', qos.RULE_TYPE_BANDWIDTH_LIMIT)] = (
self._delete_bw_limit_rule)
self.br_int = ovs_lib.OVSBridge(self.br_int_name)
def create(self, port, rules):
self._handle_rules('create', port, rules)
def update(self, port, rules):
self._handle_rules('update', port, rules)
def delete(self, port, rules):
self._handle_rules('delete', port, rules)
def _handle_rules(self, action, port, rules):
for rule in rules:
handler = self.handlers.get((action, rule.get('type')))
if handler is not None:
handler(port, rule)
def _update_bw_limit_rule(self, port, rule):
port_name = port.get('name')
max_kbps = rule.get('max_kbps')
max_burst_kbps = rule.get('max_burst_kbps')
current_max_kbps, current_max_burst = (
self.br_int.get_qos_bw_limit_for_port(port_name))
if current_max_kbps is not None or current_max_burst is not None:
self.br_int.del_qos_bw_limit_for_port(port_name)
self.br_int.create_qos_bw_limit_for_port(port_name,
max_kbps,
max_burst_kbps)
def _delete_bw_limit_rule(self, port, rule):
port_name = port.get('name')
current_max_kbps, current_max_burst = (
self.br_int.get_qos_bw_limit_for_port(port_name))
if current_max_kbps is not None or current_max_burst is not None:
self.br_int.del_qos_bw_limit_for_port(port_name)

8
neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py

@ -30,6 +30,7 @@ from six import moves
from neutron.agent.common import ovs_lib
from neutron.agent.common import polling
from neutron.agent.common import utils
from neutron.agent.l2 import agent_extensions_manager
from neutron.agent.linux import ip_lib
from neutron.agent import rpc as agent_rpc
from neutron.agent import securitygroups_rpc as sg_rpc
@ -226,6 +227,7 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
# keeps association between ports and ofports to detect ofport change
self.vifname_to_ofport_map = {}
self.setup_rpc()
self.init_agent_extensions_mgr()
self.bridge_mappings = bridge_mappings
self.setup_physical_bridges(self.bridge_mappings)
self.local_vlan_map = {}
@ -364,6 +366,11 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
consumers,
start_listening=False)
def init_agent_extensions_mgr(self):
self.agent_extensions_mgr = (
agent_extensions_manager.AgentExtensionsManager())
self.agent_extensions_mgr.initialize()
def get_net_uuid(self, vif_id):
for network_id, vlan_mapping in six.iteritems(self.local_vlan_map):
if vif_id in vlan_mapping.vif_ports:
@ -1240,6 +1247,7 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
if need_binding:
details['vif_port'] = port
need_binding_devices.append(details)
self.agent_extensions_mgr.handle_port(self.context, details)
else:
LOG.warn(_LW("Device %s not defined on plugin"), device)
if (port and port.ofport != -1):

50
neutron/plugins/ml2/extensions/qos.py Normal file

@ -0,0 +1,50 @@
# Copyright (c) 2015 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.plugins.ml2 import driver_api as api
from neutron.services.qos import qos_extension
LOG = logging.getLogger(__name__)
class QosExtensionDriver(api.ExtensionDriver):
def initialize(self):
self.qos_ext_handler = qos_extension.QosResourceExtensionHandler()
LOG.debug("QosExtensionDriver initialization complete")
def process_create_network(self, context, data, result):
self.qos_ext_handler.process_resource(
context, qos_extension.NETWORK, data, result)
process_update_network = process_create_network
def process_create_port(self, context, data, result):
self.qos_ext_handler.process_resource(
context, qos_extension.PORT, data, result)
process_update_port = process_create_port
def extend_network_dict(self, session, db_data, result):
result.update(
self.qos_ext_handler.extract_resource_fields(qos_extension.NETWORK,
db_data))
def extend_port_dict(self, session, db_data, result):
result.update(
self.qos_ext_handler.extract_resource_fields(qos_extension.PORT,
db_data))

15
neutron/plugins/ml2/managers.py

@ -723,10 +723,14 @@ class ExtensionManager(stevedore.named.NamedExtensionManager):
# the order in which the drivers are called.
self.ordered_ext_drivers = []
#TODO(QoS): enforce qos extension until we enable it in devstack-gate
drivers = cfg.CONF.ml2.extension_drivers
if 'qos' not in drivers:
drivers += ['qos']
LOG.info(_LI("Configured extension driver names: %s"),
cfg.CONF.ml2.extension_drivers)
drivers)
super(ExtensionManager, self).__init__('neutron.ml2.extension_drivers',
cfg.CONF.ml2.extension_drivers,
drivers,
invoke_on_load=True,
name_order=True)
LOG.info(_LI("Loaded extension driver names: %s"), self.names())
@ -753,9 +757,10 @@ class ExtensionManager(stevedore.named.NamedExtensionManager):
exts = []
for driver in self.ordered_ext_drivers:
alias = driver.obj.extension_alias
exts.append(alias)
LOG.info(_LI("Got %(alias)s extension from driver '%(drv)s'"),
{'alias': alias, 'drv': driver.name})
if alias:
exts.append(alias)
LOG.info(_LI("Got %(alias)s extension from driver '%(drv)s'"),
{'alias': alias, 'drv': driver.name})
return exts
def _call_on_ext_drivers(self, method_name, plugin_context, data, result):

30
neutron/plugins/ml2/plugin.py

@ -31,6 +31,7 @@ from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
from neutron.api.rpc.handlers import dhcp_rpc
from neutron.api.rpc.handlers import dvr_rpc
from neutron.api.rpc.handlers import metadata_rpc
from neutron.api.rpc.handlers import resources_rpc
from neutron.api.rpc.handlers import securitygroups_rpc
from neutron.api.v2 import attributes
from neutron.callbacks import events
@ -62,6 +63,7 @@ from neutron.extensions import extra_dhcp_opt as edo_ext
from neutron.extensions import portbindings
from neutron.extensions import portsecurity as psec
from neutron.extensions import providernet as provider
from neutron.extensions import qos
from neutron.extensions import vlantransparent
from neutron.i18n import _LE, _LI, _LW
from neutron import manager
@ -152,7 +154,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
dvr_rpc.DVRServerRpcCallback(),
dhcp_rpc.DhcpRpcCallback(),
agents_db.AgentExtRpcCallback(),
metadata_rpc.MetadataRpcCallback()
metadata_rpc.MetadataRpcCallback(),
resources_rpc.ResourcesServerRpcCallback()
]
def _setup_dhcp(self):
@ -614,6 +617,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
def create_network(self, context, network):
result, mech_context = self._create_network_with_retries(context,
network)
self._notify_registry(
resources.NETWORK, events.AFTER_CREATE, context, result)
try:
self.mechanism_manager.create_network_postcommit(mech_context)
except ml2_exc.MechanismDriverError:
@ -626,6 +631,12 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
def create_network_bulk(self, context, networks):
objects = self._create_bulk_ml2(attributes.NETWORK, context, networks)
for obj in objects:
self._notify_registry(resources.NETWORK,
events.AFTER_CREATE,
context,
obj)
return [obj['result'] for obj in objects]
def update_network(self, context, id, network):
@ -648,6 +659,10 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
original_network=original_network)
self.mechanism_manager.update_network_precommit(mech_context)
# Notifications must be sent after the above transaction is complete
self._notify_registry(
resources.NETWORK, events.AFTER_UPDATE, context, updated_network)
# TODO(apech) - handle errors raised by update_network, potentially
# by re-calling update_network with the previous attributes. For
# now the error is propogated to the caller, which is expected to
@ -1110,6 +1125,12 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
original_port[psec.PORTSECURITY] !=
updated_port[psec.PORTSECURITY]):
need_port_update_notify = True
# TODO(QoS): Move out to the extension framework somehow.
# Follow https://review.openstack.org/#/c/169223 for a solution.
if (qos.QOS_POLICY_ID in attrs and
original_port[qos.QOS_POLICY_ID] !=
updated_port[qos.QOS_POLICY_ID]):
need_port_update_notify = True
if addr_pair.ADDRESS_PAIRS in attrs:
need_port_update_notify |= (
@ -1510,3 +1531,10 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
if port:
return port.id
return device
def _notify_registry(self, resource_type, event_type, context, resource):
kwargs = {
'context': context,
resource_type: resource,
}
registry.notify(resource_type, event_type, self, **kwargs)

0
neutron/services/qos/__init__.py Normal file

82
neutron/services/qos/qos_extension.py Normal file

@ -0,0 +1,82 @@
# Copyright (c) 2015 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.extensions import qos
from neutron import manager
from neutron.objects.qos import policy as policy_object
from neutron.plugins.common import constants as plugin_constants
NETWORK = 'network'
PORT = 'port'
# TODO(QoS): Add interface to define how this should look like
class QosResourceExtensionHandler(object):
@property
def plugin_loaded(self):
if not hasattr(self, '_plugin_loaded'):
service_plugins = manager.NeutronManager.get_service_plugins()
self._plugin_loaded = plugin_constants.QOS in service_plugins
return self._plugin_loaded
def _get_policy_obj(self, context, policy_id):
return policy_object.QosPolicy.get_by_id(context, policy_id)
def _update_port_policy(self, context, port, port_changes):
old_policy = policy_object.QosPolicy.get_port_policy(
context, port['id'])
if old_policy:
#TODO(QoS): this means two transactions. One for detaching
# one for re-attaching, we may want to update
# within a single transaction instead, or put
# a whole transaction on top, or handle the switch
# at db api level automatically within transaction.
old_policy.detach_port(port['id'])
qos_policy_id = port_changes.get(qos.QOS_POLICY_ID)
if qos_policy_id is not None:
policy = self._get_policy_obj(context, qos_policy_id)
policy.attach_port(port['id'])
port[qos.QOS_POLICY_ID] = qos_policy_id
def _update_network_policy(self, context, network, network_changes):
old_policy = policy_object.QosPolicy.get_network_policy(
context, network['id'])
if old_policy:
old_policy.detach_network(network['id'])
qos_policy_id = network_changes.get(qos.QOS_POLICY_ID)
if qos_policy_id:
policy = self._get_policy_obj(context, qos_policy_id)
policy.attach_network(network['id'])
network[qos.QOS_POLICY_ID] = qos_policy_id
def _exec(self, method_name, context, kwargs):
return getattr(self, method_name)(context=context, **kwargs)
def process_resource(self, context, resource_type, requested_resource,
actual_resource):
if qos.QOS_POLICY_ID in requested_resource and self.plugin_loaded:
self._exec('_update_%s_policy' % resource_type, context,
{resource_type: actual_resource,
"%s_changes" % resource_type: requested_resource})
def extract_resource_fields(self, resource_type, resource):
if not self.plugin_loaded:
return {}
binding = resource['qos_policy_binding']
return {qos.QOS_POLICY_ID: binding['policy_id'] if binding else None}

194
neutron/services/qos/qos_plugin.py Normal file

@ -0,0 +1,194 @@
# Copyright (c) 2015 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 import manager
from neutron.api.rpc.callbacks import registry as rpc_registry
from neutron.api.rpc.callbacks import resources as rpc_resources
from neutron.extensions import qos
from neutron.i18n import _LW
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.plugins.common import constants
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
#TODO(QoS): remove this stub when db is ready
def _get_qos_policy_cb_stub(resource, policy_id, **kwargs):
"""Hardcoded stub for testing until we get the db working."""
qos_policy = {
"tenant_id": "8d4c70a21fed4aeba121a1a429ba0d04",
"id": "46ebaec0-0570-43ac-82f6-60d2b03168c4",
"name": "10Mbit",
"description": "This policy limits the ports to 10Mbit max.",
"shared": False,
"rules": [{
"id": "5f126d84-551a-4dcf-bb01-0e9c0df0c793",
"max_kbps": "10000",
"max_burst_kbps": "0",
"type": "bandwidth_limit"
}]
}
return qos_policy
def _get_qos_policy_cb(resource, policy_id, **kwargs):
qos_plugin = manager.NeutronManager.get_service_plugins().get(
constants.QOS)
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
qos_policy = qos_plugin.get_qos_policy(context, policy_id)
return qos_policy
#TODO(QoS): remove this stub when db is ready
def _get_qos_bandwidth_limit_rule_cb_stub(resource, rule_id, **kwargs):
"""Hardcoded for testing until we get the db working."""
bandwidth_limit = {
"id": "5f126d84-551a-4dcf-bb01-0e9c0df0c793",
"qos_policy_id": "46ebaec0-0570-43ac-82f6-60d2b03168c4",
"max_kbps": "10000",
"max_burst_kbps": "0",
}
return bandwidth_limit
def _get_qos_bandwidth_limit_rule_cb(resource, rule_id, **kwargs):
qos_plugin = manager.NeutronManager.get_service_plugins().get(
constants.QOS)
context = kwargs.get('context')
if context is None:
LOG.warning(_LW(
'Received %(resource)s %(rule_id,)s without context '),
{'resource': resource, 'rule_id,': rule_id}
)
return
bandwidth_limit = qos_plugin.get_qos_bandwidth_limit_rule(
context,
rule_id)
return bandwidth_limit
class QoSPlugin(qos.QoSPluginBase):
"""Implementation of the Neutron QoS Service Plugin.
This class implements a Quality of Service plugin that
provides quality of service parameters over ports and
networks.
"""
supported_extension_aliases = ['qos']
def __init__(self):
super(QoSPlugin, self).__init__()
self.register_resource_providers()
def register_resource_providers(self):
rpc_registry.register_provider(
_get_qos_bandwidth_limit_rule_cb_stub,
rpc_resources.QOS_RULE)
rpc_registry.register_provider(
_get_qos_policy_cb_stub,
rpc_resources.QOS_POLICY)
def create_policy(self, context, policy):
policy = policy_object.QosPolicy(context, **policy['policy'])
policy.create()
return policy.to_dict()
def update_policy(self, context, policy_id, policy):
policy = policy_object.QosPolicy(context, **policy['policy'])
policy.id = policy_id
policy.update()
return policy.to_dict()
def delete_policy(self, context, policy_id):
policy = policy_object.QosPolicy(context)
policy.id = policy_id
policy.delete()
def _get_policy_obj(self, context, policy_id):
return policy_object.QosPolicy.get_by_id(context, policy_id)
def get_policy(self, context, policy_id, fields=None):
#TODO(QoS): Support the fields parameter
return self._get_policy_obj(context, policy_id).to_dict()
def get_policies(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
#TODO(QoS): Support all the optional parameters
return [policy_obj.to_dict() for policy_obj in
policy_object.QosPolicy.get_objects(context)]
#TODO(QoS): Consider adding a proxy catch-all for rules, so
# we capture the API function call, and just pass
# the rule type as a parameter removing lots of
# future code duplication when we have more rules.
def create_policy_bandwidth_limit_rule(self, context, policy_id,
bandwidth_limit_rule):
#TODO(QoS): avoid creation of severan bandwidth limit rules
# in the future we need an inter-rule validation
# mechanism to verify all created rules will
# play well together.
rule = rule_object.QosBandwidthLimitRule(
context, qos_policy_id=policy_id,
**bandwidth_limit_rule['bandwidth_limit_rule'])
rule.create()
return rule.to_dict()
def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id,
bandwidth_limit_rule):
rule = rule_object.QosBandwidthLimitRule(
context, **bandwidth_limit_rule['bandwidth_limit_rule'])
rule.id = rule_id
rule.update()
return rule.to_dict()
def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id):
rule = rule_object.QosBandwidthLimitRule()
rule.id = rule_id
rule.delete()
def get_policy_bandwidth_limit_rule(self, context, rule_id,
policy_id, fields=None):
#TODO(QoS): Support the fields parameter
return rule_object.QosBandwidthLimitRule.get_by_id(context,
rule_id).to_dict()
def get_policy_bandwidth_limit_rules(self, context, policy_id,
filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
#TODO(QoS): Support all the optional parameters
return [rule_obj.to_dict() for rule_obj in
rule_object.QosBandwidthLimitRule.get_objects(context)]
def get_rule_types(self, context, filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
pass

28
neutron/tests/api/base.py

@ -88,6 +88,8 @@ class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
cls.fw_rules = []
cls.fw_policies = []
cls.ipsecpolicies = []
cls.qos_rules = []
cls.qos_policies = []
cls.ethertype = "IPv" + str(cls._ip_version)
cls.address_scopes = []
cls.admin_address_scopes = []
@ -107,6 +109,14 @@ class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
for fw_rule in cls.fw_rules:
cls._try_delete_resource(cls.client.delete_firewall_rule,
fw_rule['id'])
# Clean up QoS policies
for qos_policy in cls.qos_policies:
cls._try_delete_resource(cls.client.delete_qos_policy,
qos_policy['id'])
# Clean up QoS rules
for qos_rule in cls.qos_rules:
cls._try_delete_resource(cls.client.delete_qos_rule,
qos_rule['id'])
# Clean up ike policies
for ikepolicy in cls.ikepolicies:
cls._try_delete_resource(cls.client.delete_ikepolicy,
@ -431,6 +441,24 @@ class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
cls.fw_policies.append(fw_policy)
return fw_policy
@classmethod
def create_qos_policy(cls, name, description, shared):
"""Wrapper utility that returns a test QoS policy."""
body = cls.client.create_qos_policy(name, description, shared)
qos_policy = body['policy']
cls.qos_policies.append(qos_policy)
return qos_policy
@classmethod
def create_qos_bandwidth_limit_rule(cls, policy_id,
max_kbps, max_burst_kbps):
"""Wrapper utility that returns a test QoS bandwidth limit rule."""
body = cls.client.create_bandwidth_limit_rule(
policy_id, max_kbps, max_burst_kbps)
qos_rule = body['bandwidth_limit_rule']
cls.qos_rules.append(qos_rule)
return qos_rule
@classmethod
def delete_router(cls, router):
body = cls.client.list_router_interfaces(router['id'])

78
neutron/tests/api/test_qos.py Normal file

@ -0,0 +1,78 @@
# Copyright 2015 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from neutron.tests.api import base
from neutron.tests.tempest import config
from neutron.tests.tempest import test
CONF = config.CONF
class QosTestJSON(base.BaseAdminNetworkTest):
@classmethod
def resource_setup(cls):
super(QosTestJSON, cls).resource_setup()
if not test.is_extension_enabled('qos', 'network'):
msg = "qos extension not enabled."
raise cls.skipException(msg)
@test.attr(type='smoke')
@test.idempotent_id('108fbdf7-3463-4e47-9871-d07f3dcf5bbb')
def test_create_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy desc',
shared=False)
# Test 'show policy'
retrieved_policy = self.admin_client.show_qos_policy(policy['id'])
retrieved_policy = retrieved_policy['policy']
self.assertEqual('test-policy', retrieved_policy['name'])
self.assertEqual('test policy desc', retrieved_policy['description'])
self.assertEqual(False, retrieved_policy['shared'])
# Test 'list policies'
policies = self.admin_client.list_qos_policies()['policies']
policies_ids = [p['id'] for p in policies]
self.assertIn(policy['id'], policies_ids)
@test.attr(type='smoke')
@test.idempotent_id('8a59b00b-3e9c-4787-92f8-93a5cdf5e378')
def test_create_rule(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
rule = self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],
max_kbps=200,
max_burst_kbps=1337)
# Test 'show rule'
retrieved_policy = self.admin_client.show_bandwidth_limit_rule(
policy['id'], rule['id'])
retrieved_policy = retrieved_policy['bandwidth_limit_rule']
self.assertEqual(rule['id'], retrieved_policy['id'])
self.assertEqual(200, retrieved_policy['max_kbps'])
self.assertEqual(1337, retrieved_policy['max_burst_kbps'])
# Test 'list rules'
rules = self.admin_client.list_bandwidth_limit_rules(policy['id'])
rules = rules['bandwidth_limit_rules']
rules_ids = [r['id'] for r in rules]
self.assertIn(rule['id'], rules_ids)
#TODO(QoS): policy update (name)
#TODO(QoS): create several bandwidth-limit rules (not sure it makes sense,
# but to test more than one rule)
#TODO(QoS): update bandwidth-limit rule
#TODO(QoS): associate/disassociate policy with network
#TODO(QoS): associate/disassociate policy with port

6
neutron/tests/etc/policy.json

@ -39,12 +39,14 @@
"get_network:provider:physical_network": "rule:admin_only",
"get_network:provider:segmentation_id": "rule:admin_only",
"get_network:queue_id": "rule:admin_only",
"get_network:qos_policy_id": "rule:admin_only",
"create_network:shared": "rule:admin_only",
"create_network:router:external": "rule:admin_only",
"create_network:segments": "rule:admin_only",
"create_network:provider:network_type": "rule:admin_only",
"create_network:provider:physical_network": "rule:admin_only",
"create_network:provider:segmentation_id": "rule:admin_only",
"create_network:qos_policy_id": "rule:admin_only",
"update_network": "rule:admin_or_owner",
"update_network:segments": "rule:admin_only",
"update_network:shared": "rule:admin_only",
@ -52,6 +54,7 @@
"update_network:provider:physical_network": "rule:admin_only",
"update_network:provider:segmentation_id": "rule:admin_only",
"update_network:router:external": "rule:admin_only",
"update_network:qos_policy_id": "rule:admin_only",
"delete_network": "rule:admin_or_owner",
"create_port": "",
@ -62,12 +65,14 @@
"create_port:binding:profile": "rule:admin_only",
"create_port:mac_learning_enabled": "rule:admin_or_network_owner or rule:context_is_advsvc",
"create_port:allowed_address_pairs": "rule:admin_or_network_owner",
"create_port:qos_policy_id": "rule:admin_only",
"get_port": "rule:admin_or_owner or rule:context_is_advsvc",
"get_port:queue_id": "rule:admin_only",
"get_port:binding:vif_type": "rule:admin_only",
"get_port:binding:vif_details": "rule:admin_only",
"get_port:binding:host_id": "rule:admin_only",
"get_port:binding:profile": "rule:admin_only",
"get_port:qos_policy_id": "rule:admin_only",
"update_port": "rule:admin_or_owner or rule:context_is_advsvc",
"update_port:mac_address": "rule:admin_only or rule:context_is_advsvc",
"update_port:fixed_ips": "rule:admin_or_network_owner or rule:context_is_advsvc",
@ -76,6 +81,7 @@
"update_port:binding:profile": "rule:admin_only",
"update_port:mac_learning_enabled": "rule:admin_or_network_owner or rule:context_is_advsvc",
"update_port:allowed_address_pairs": "rule:admin_or_network_owner",
"update_port:qos_policy_id": "rule:admin_only",
"delete_port": "rule:admin_or_owner or rule:context_is_advsvc",
"get_router:ha": "rule:admin_only",

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

@ -288,6 +288,17 @@ class OVSBridgeTestCase(OVSBridgeTestBase):
controller,
'connection_mode'))
def test_qos_bw_limit(self):
port_name, _ = self.create_ovs_port()
self.br.create_qos_bw_limit_for_port(port_name, 700, 70)
max_rate, burst = self.br.get_qos_bw_limit_for_port(port_name)
self.assertEqual(700, max_rate)
self.assertEqual(70, burst)
self.br.del_qos_bw_limit_for_port(port_name)
max_rate, burst = self.br.get_qos_bw_limit_for_port(port_name)
self.assertIsNone(max_rate)
self.assertIsNone(burst)
class OVSLibTestCase(base.BaseOVSLinuxTestCase):

76
neutron/tests/tempest/services/network/json/network_client.py

@ -65,7 +65,9 @@ class NetworkClientJSON(service_client.ServiceClient):
'metering_label_rules': 'metering',
'firewall_rules': 'fw',
'firewall_policies': 'fw',
'firewalls': 'fw'
'firewalls': 'fw',
'policies': 'qos',
'bandwidth_limit_rules': 'qos',
}
service_prefix = service_resource_prefix_map.get(
plural_name)
@ -90,7 +92,8 @@ class NetworkClientJSON(service_client.ServiceClient):
'ikepolicy': 'ikepolicies',
'ipsec_site_connection': 'ipsec-site-connections',
'quotas': 'quotas',
'firewall_policy': 'firewall_policies'
'firewall_policy': 'firewall_policies',
'qos_policy': 'policies'
}
return resource_plural_map.get(resource_name, resource_name + 's')
@ -620,3 +623,72 @@ class NetworkClientJSON(service_client.ServiceClient):
self.expected_success(200, resp.status)
body = json.loads(body)
return service_client.ResponseBody(resp, body)
def list_qos_policies(self):
uri = '%s/qos/policies' % self.uri_prefix
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = json.loads(body)
return service_client.ResponseBody(resp, body)
def create_qos_policy(self, name, description, shared):
uri = '%s/qos/policies' % self.uri_prefix
post_data = self.serialize(
{'policy': {
'name': name,
'description': description,
'shared': shared
}})
resp, body = self.post(uri, post_data)
body = self.deserialize_single(body)
self.expected_success(201, resp.status)
return service_client.ResponseBody(resp, body)
def get_qos_policy(self, policy_id):
uri = '%s/qos/policies/%s' % (self.uri_prefix, policy_id)
resp, body = self.get(uri)
self.expected_success(200, resp.status)
return service_client.ResponseBody(resp, body)
def create_bandwidth_limit_rule(self, policy_id, max_kbps, max_burst_kbps):
uri = '%s/qos/policies/%s/bandwidth_limit_rules' % (
self.uri_prefix, policy_id)
#TODO(QoS): 'bandwidth_limit' should not be a magic string.
post_data = self.serialize(
{'bandwidth_limit_rule': {
'max_kbps': max_kbps,
'max_burst_kbps': max_burst_kbps,
'type': 'bandwidth_limit'}})
resp, body = self.post(uri, post_data)
self.expected_success(201, resp.status)
body = json.loads(body)
return service_client.ResponseBody(resp, body)
def list_bandwidth_limit_rules(self, policy_id):
uri = '%s/qos/policies/%s/bandwidth_limit_rules' % (
self.uri_prefix, policy_id)
resp, body = self.get(uri)
body = self.deserialize_single(body)
self.expected_success(200, resp.status)
return service_client.ResponseBody(resp, body)
def show_bandwidth_limit_rule(self, policy_id, rule_id):
uri = '%s/qos/policies/%s/bandwidth_limit_rules/%s' % (
self.uri_prefix, policy_id, rule_id)
resp, body = self.get(uri)
body = self.deserialize_single(body)
self.expected_success(200, resp.status)
return service_client.ResponseBody(resp, body)
def update_bandwidth_limit_rule(self, policy_id, rule_id,
max_kbps, max_burst_kbps):
uri = '%s/qos/policies/%s/bandwidth_limit_rules/%s' % (
self.uri_prefix, policy_id, rule_id)
post_data = {
'bandwidth_limit_rule': {
'max_kbps': max_kbps,
'max_burst_kbps': max_burst_kbps,
'type': 'bandwidth_limit'}}
resp, body = self.put(uri, json.dumps(post_data))
self.expected_success(200, resp.status)
return service_client.ResponseBody(resp, body)

0
neutron/tests/unit/agent/l2/__init__.py Executable file

0
neutron/tests/unit/agent/l2/extensions/__init__.py Executable file

91
neutron/tests/unit/agent/l2/extensions/test_qos_agent.py Executable file

@ -0,0 +1,91 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
# 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.
import mock
from oslo_utils import uuidutils
from neutron.agent.l2.extensions import qos_agent
from neutron.api.rpc.callbacks import resources
from neutron import context
from neutron.tests import base
# This is a minimalistic mock of rules to be passed/checked around
# which should be exteneded as needed to make real rules
TEST_GET_INFO_RULES = ['rule1', 'rule2']
class QosAgentExtensionTestCase(base.BaseTestCase):
def setUp(self):
super(QosAgentExtensionTestCase, self).setUp()
self.qos_agent = qos_agent.QosAgentExtension()
self.context = context.get_admin_context()
# Don't rely on used driver
mock.patch(
'neutron.manager.NeutronManager.load_class_for_provider',
return_value=lambda: mock.Mock(spec=qos_agent.QosAgentDriver)
).start()
self.qos_agent.initialize()
self._create_fake_resource_rpc()
def _create_fake_resource_rpc(self):
self.get_info_mock = mock.Mock(return_value=TEST_GET_INFO_RULES)
self.qos_agent.resource_rpc.get_info = self.get_info_mock
def _create_test_port_dict(self):
return {'port_id': uuidutils.generate_uuid(),
'qos_policy_id': uuidutils.generate_uuid()}
def test_handle_port_with_no_policy(self):
port = self._create_test_port_dict()
del port['qos_policy_id']
self.qos_agent._process_rules_updates = mock.Mock()
self.qos_agent.handle_port(self.context, port)
self.assertFalse(self.qos_agent._process_rules_updates.called)
def test_handle_unknown_port(self):
port = self._create_test_port_dict()
qos_policy_id = port['qos_policy_id']
port_id = port['port_id']
self.qos_agent.handle_port(self.context, port)
# we make sure the underlaying qos driver is called with the
# right parameters
self.qos_agent.qos_driver.create.assert_called_once_with(
port, TEST_GET_INFO_RULES)
self.assertEqual(port,
self.qos_agent.qos_policy_ports[qos_policy_id][port_id])
self.assertTrue(port_id in self.qos_agent.known_ports)
def test_handle_known_port(self):
port_obj1 = self._create_test_port_dict()
port_obj2 = dict(port_obj1)
self.qos_agent.handle_port(self.context, port_obj1)
self.qos_agent.qos_driver.reset_mock()
self.qos_agent.handle_port(self.context, port_obj2)
self.assertFalse(self.qos_agent.qos_driver.create.called)
def test_handle_known_port_change_policy_id(self):
port = self._create_test_port_dict()
self.qos_agent.handle_port(self.context, port)
self.qos_agent.resource_rpc.get_info.reset_mock()
port['qos_policy_id'] = uuidutils.generate_uuid()
self.qos_agent.handle_port(self.context, port)
self.get_info_mock.assert_called_once_with(
self.context, resources.QOS_POLICY,
port['qos_policy_id'])
#TODO(QoS): handle qos_driver.update call check when
# we do that

0
neutron/tests/unit/api/rpc/callbacks/__init__.py Normal file

78
neutron/tests/unit/api/rpc/callbacks/test_resource_manager.py Normal file

@ -0,0 +1,78 @@
# 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.api.rpc.callbacks import registry as rpc_registry
from neutron.api.rpc.callbacks import resources
from neutron.tests import base
class ResourcesCallbackRequestTestCase(base.BaseTestCase):
def setUp(self):
super(ResourcesCallbackRequestTestCase, self).setUp()
self.resource_id = '46ebaec0-0570-43ac-82f6-60d2b03168c4'
self.qos_rule_id = '5f126d84-551a-4dcf-bb01-0e9c0df0c793'
def test_resource_callback_request(self):
#TODO(QoS) convert it to the version object format
def _get_qos_policy_cb(resource, policy_id, **kwargs):
qos_policy = {
"tenant_id": "8d4c70a21fed4aeba121a1a429ba0d04",
"id": "46ebaec0-0570-43ac-82f6-60d2b03168c4",
"name": "10Mbit",
"description": "This policy limits the ports to 10Mbit max.",
"shared": False,
"rules": [{
"id": "5f126d84-551a-4dcf-bb01-0e9c0df0c793",
"max_kbps": "10000",
"max_burst_kbps": "0",
"type": "bnadwidth_limit"
}]
}
return qos_policy
#TODO(QoS) convert it to the version object format
def _get_qos_bandwidth_limit_rule_cb(resource, rule_id, **kwargs):
bandwidth_limit = {
"id": "5f126d84-551a-4dcf-bb01-0e9c0df0c793",
"qos_policy_id": "46ebaec0-0570-43ac-82f6-60d2b03168c4",
"max_kbps": "10000",
"max_burst_kbps": "0",
}
return bandwidth_limit
rpc_registry.register_provider(
_get_qos_bandwidth_limit_rule_cb,
resources.QOS_RULE)
rpc_registry.register_provider(
_get_qos_policy_cb,
resources.QOS_POLICY)
self.ctx = None
kwargs = {'context': self.ctx}
qos_policy = rpc_registry.get_info(
resources.QOS_POLICY,
self.resource_id,
**kwargs)
self.assertEqual(self.resource_id, qos_policy['id'])
qos_rule = rpc_registry.get_info(
resources.QOS_RULE,
self.qos_rule_id,
**kwargs)
self.assertEqual(self.qos_rule_id, qos_rule['id'])

11
neutron/tests/unit/common/test_utils.py

@ -679,3 +679,14 @@ class TestEnsureDir(base.BaseTestCase):
def test_ensure_dir_calls_makedirs(self, makedirs):
utils.ensure_dir("/etc/create/directory")
makedirs.assert_called_once_with("/etc/create/directory", 0o755)
class TestCamelize(base.BaseTestCase):
def test_camelize(self):
data = {'bandwidth_limit': 'BandwidthLimit',
'test': 'Test',
'some__more__dashes': 'SomeMoreDashes',
'a_penguin_walks_into_a_bar': 'APenguinWalksIntoABar'}
for s, expected in data.items():
self.assertEqual(expected, utils.camelize(s))

0
neutron/tests/unit/objects/__init__.py Normal file

0
neutron/tests/unit/objects/qos/__init__.py Normal file

126
neutron/tests/unit/objects/qos/test_policy.py Normal file

@ -0,0 +1,126 @@
# 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 api as db_api
from neutron.db import models_v2
from neutron.objects.qos import policy
from neutron.objects.qos import rule
from neutron.tests.unit.objects import test_base
from neutron.tests.unit import testlib_api
class QosPolicyObjectTestCase(test_base.BaseObjectIfaceTestCase):
_test_class = policy.QosPolicy
class QosPolicyDbObjectTestCase(test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = policy.QosPolicy
def setUp(self):
super(QosPolicyDbObjectTestCase, self).setUp()
self._create_test_network()
self._create_test_port(self._network)
#TODO(QoS): move _create_test_policy here, as it's common
# to all. Now the base DB Object test case breaks
# that by introducing a duplicate object colliding
# on PK.
def _create_test_policy(self):
policy_obj = policy.QosPolicy(self.context, **self.db_obj)
policy_obj.create()
return policy_obj
def _create_test_network(self):
# TODO(ihrachys): replace with network.create() once we get an object
# implementation for networks
self._network = db_api.create_object(self.context, models_v2.Network,
{'name': 'test-network1'})
def _create_test_port(self, network):
# TODO(ihrachys): replace with port.create() once we get an object
# implementation for ports
self._port = db_api.create_object(self.context, models_v2.Port,
{'name': 'test-port1',
'network_id': network['id'],
'mac_address': 'fake_mac',
'admin_state_up': True,
'status': 'ACTIVE',
'device_id': 'fake_device',
'device_owner': 'fake_owner'})
#TODO(QoS): give a thought on checking detach/attach for invalid values.
def test_attach_network_get_network_policy(self):
obj = self._create_test_policy()
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self.assertIsNone(policy_obj)
# Now attach policy and repeat
obj.attach_network(self._network['id'])
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self.assertEqual(obj, policy_obj)
def test_attach_port_get_port_policy(self):
obj = self._create_test_policy()
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self.assertIsNone(policy_obj)
# Now attach policy and repeat
obj.attach_port(self._port['id'])
policy_obj = policy.QosPolicy.get_port_policy(self.context,
self._port['id'])
self.assertEqual(obj, policy_obj)
def test_detach_port(self):
obj = self._create_test_policy()
obj.attach_port(self._port['id'])
obj.detach_port(self._port['id'])
policy_obj = policy.QosPolicy.get_port_policy(self.context,
self._port['id'])
self.assertIsNone(policy_obj)
def test_detach_network(self):
obj = self._create_test_policy()
obj.attach_network(self._network['id'])
obj.detach_network(self._network['id'])
policy_obj = policy.QosPolicy.get_network_policy(self.context,
self._network['id'])
self.assertIsNone(policy_obj)
def test_synthetic_rule_fields(self):
obj = policy.QosPolicy(self.context, **self.db_obj)
obj.create()
rule_fields = self.get_random_fields(
obj_cls=rule.QosBandwidthLimitRule)
rule_fields['qos_policy_id'] = obj.id
rule_fields['tenant_id'] = obj.tenant_id
rule_obj = rule.QosBandwidthLimitRule(self.context, **rule_fields)
rule_obj.create()
obj = policy.QosPolicy.get_by_id(self.context, obj.id)
self.assertEqual([rule_obj], obj.bandwidth_limit_rules)

129
neutron/tests/unit/objects/qos/test_rule.py Normal file

@ -0,0 +1,129 @@
# 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.db import api as db_api
from neutron.objects.qos import policy
from neutron.objects.qos import rule
from neutron.tests.unit.objects import test_base
from neutron.tests.unit import testlib_api
class QosBandwidthLimitRuleObjectTestCase(test_base.BaseObjectIfaceTestCase):
_test_class = rule.QosBandwidthLimitRule
@classmethod
def get_random_fields(cls):
# object middleware should not allow random types, so override it with
# proper type
fields = (super(QosBandwidthLimitRuleObjectTestCase, cls)
.get_random_fields())
fields['type'] = cls._test_class.rule_type
return fields
def _filter_db_object(self, func):
return {
field: self.db_obj[field]
for field in self._test_class.fields
if func(field)
}
def _get_core_db_obj(self):
return self._filter_db_object(
lambda field: self._test_class._is_core_field(field))
def _get_addn_db_obj(self):
return self._filter_db_object(
lambda field: self._test_class._is_addn_field(field))
def test_get_by_id(self):
with mock.patch.object(db_api, 'get_object',
return_value=self.db_obj) as get_object_mock:
obj = self._test_class.get_by_id(self.context, id='fake_id')
self.assertTrue(self._is_test_class(obj))
self.assertEqual(self.db_obj, test_base.get_obj_db_fields(obj))
get_object_mock.assert_has_calls([
mock.call(self.context, model, id='fake_id')
for model in (self._test_class.db_model,
self._test_class.base_db_model)
], any_order=True)
def test_get_objects(self):
with mock.patch.object(db_api, 'get_objects',
return_value=self.db_objs):
@classmethod
def _get_by_id(cls, context, id):
for db_obj in self.db_objs:
if db_obj['id'] == id:
return self._test_class(context, **db_obj)
with mock.patch.object(rule.QosRule, 'get_by_id', new=_get_by_id):
objs = self._test_class.get_objects(self.context)
self.assertFalse(
filter(lambda obj: not self._is_test_class(obj), objs))
self.assertEqual(
sorted(self.db_objs),
sorted(test_base.get_obj_db_fields(obj) for obj in objs))
def test_create(self):
with mock.patch.object(db_api, 'create_object',
return_value=self.db_obj) as create_mock:
test_class = self._test_class
obj = test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.create()
self._check_equal(obj, self.db_obj)
core_db_obj = self._get_core_db_obj()
addn_db_obj = self._get_addn_db_obj()
create_mock.assert_has_calls(
[mock.call(self.context, self._test_class.base_db_model,
core_db_obj),
mock.call(self.context, self._test_class.db_model,
addn_db_obj)]
)
def test_update_changes(self):
with mock.patch.object(db_api, 'update_object',
return_value=self.db_obj) as update_mock:
obj = self._test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.update()
self._check_equal(obj, self.db_obj)
core_db_obj = self._get_core_db_obj()
update_mock.assert_any_call(
self.context, self._test_class.base_db_model, obj.id,
core_db_obj)
addn_db_obj = self._get_addn_db_obj()
update_mock.assert_any_call(
self.context, self._test_class.db_model, obj.id,
addn_db_obj)
class QosBandwidthLimitRuleDbObjectTestCase(test_base.BaseDbObjectTestCase,
testlib_api.SqlTestCase):
_test_class = rule.QosBandwidthLimitRule
def setUp(self):
super(QosBandwidthLimitRuleDbObjectTestCase, self).setUp()
# Prepare policy to be able to insert a rule
generated_qos_policy_id = self.db_obj['qos_policy_id']
policy_obj = policy.QosPolicy(self.context,
id=generated_qos_policy_id)
policy_obj.create()

205
neutron/tests/unit/objects/test_base.py Normal file

@ -0,0 +1,205 @@
# 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 string
import mock
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
from neutron import context
from neutron.db import api as db_api
from neutron.objects import base
from neutron.tests import base as test_base
@obj_base.VersionedObjectRegistry.register
class FakeNeutronObject(base.NeutronObject):
db_model = 'fake_model'
fields = {
'id': obj_fields.UUIDField(),
'field1': obj_fields.StringField(),
'field2': obj_fields.StringField()
}
def _random_string(n=10):
return ''.join(random.choice(string.ascii_lowercase) for _ in range(n))
def _random_boolean():
return bool(random.getrandbits(1))
def _random_integer():
return random.randint(0, 1000)
FIELD_TYPE_VALUE_GENERATOR_MAP = {
obj_fields.BooleanField: _random_boolean,
obj_fields.IntegerField: _random_integer,
obj_fields.StringField: _random_string,
obj_fields.UUIDField: _random_string,
obj_fields.ListOfObjectsField: lambda: []
}
def get_obj_db_fields(obj):
return {field: getattr(obj, field) for field in obj.fields
if field not in obj.synthetic_fields}
class _BaseObjectTestCase(object):
_test_class = FakeNeutronObject
def setUp(self):
super(_BaseObjectTestCase, self).setUp()
self.context = context.get_admin_context()
self.db_objs = list(self.get_random_fields() for _ in range(3))
self.db_obj = self.db_objs[0]
@classmethod
def get_random_fields(cls, obj_cls=None):
obj_cls = obj_cls or cls._test_class
fields = {}
for field, field_obj in obj_cls.fields.items():
if field not in obj_cls.synthetic_fields:
generator = FIELD_TYPE_VALUE_GENERATOR_MAP[type(field_obj)]
fields[field] = generator()
return fields
@classmethod
def _is_test_class(cls, obj):
return isinstance(obj, cls._test_class)
class BaseObjectIfaceTestCase(_BaseObjectTestCase, test_base.BaseTestCase):
def test_get_by_id(self):
with mock.patch.object(db_api, 'get_object',
return_value=self.db_obj) as get_object_mock:
obj = self._test_class.get_by_id(self.context, id='fake_id')
self.assertTrue(self._is_test_class(obj))
self.assertEqual(self.db_obj, get_obj_db_fields(obj))
get_object_mock.assert_called_once_with(
self.context, self._test_class.db_model, id='fake_id')
def test_get_by_id_missing_object(self):
with mock.patch.object(db_api, 'get_object', return_value=None):
obj = self._test_class.get_by_id(self.context, id='fake_id')
self.assertIsNone(obj)
def test_get_objects(self):
with mock.patch.object(db_api, 'get_objects',
return_value=self.db_objs) as get_objects_mock:
objs = self._test_class.get_objects(self.context)
self.assertFalse(
filter(lambda obj: not self._is_test_class(obj), objs))
self.assertEqual(
sorted(self.db_objs),
sorted(get_obj_db_fields(obj) for obj in objs))
get_objects_mock.assert_called_once_with(
self.context, self._test_class.db_model)
def _check_equal(self, obj, db_obj):
self.assertEqual(
sorted(db_obj),
sorted(get_obj_db_fields(obj)))
def test_create(self):
with mock.patch.object(db_api, 'create_object',
return_value=self.db_obj) as create_mock:
obj = self._test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.create()
self._check_equal(obj, self.db_obj)
create_mock.assert_called_once_with(
self.context, self._test_class.db_model, self.db_obj)
def test_create_updates_from_db_object(self):
with mock.patch.object(db_api, 'create_object',
return_value=self.db_obj):
obj = self._test_class(self.context, **self.db_objs[1])
self._check_equal(obj, self.db_objs[1])
obj.create()
self._check_equal(obj, self.db_obj)
def test_update_no_changes(self):
with mock.patch.object(db_api, 'update_object',
return_value=self.db_obj) as update_mock:
obj = self._test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.update()
self.assertTrue(update_mock.called)
# consequent call to update does not try to update database
update_mock.reset_mock()
obj.update()
self._check_equal(obj, self.db_obj)
self.assertFalse(update_mock.called)
def test_update_changes(self):
with mock.patch.object(db_api, 'update_object',
return_value=self.db_obj) as update_mock:
obj = self._test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.update()
self._check_equal(obj, self.db_obj)
update_mock.assert_called_once_with(
self.context, self._test_class.db_model,
self.db_obj['id'], self.db_obj)
def test_update_updates_from_db_object(self):
with mock.patch.object(db_api, 'update_object',
return_value=self.db_obj):
obj = self._test_class(self.context, **self.db_objs[1])
self._check_equal(obj, self.db_objs[1])
obj.update()
self._check_equal(obj, self.db_obj)
@mock.patch.object(db_api, 'delete_object')
def test_delete(self, delete_mock):
obj = self._test_class(self.context, **self.db_obj)
self._check_equal(obj, self.db_obj)
obj.delete()
self._check_equal(obj, self.db_obj)
delete_mock.assert_called_once_with(
self.context, self._test_class.db_model, self.db_obj['id'])
class BaseDbObjectTestCase(_BaseObjectTestCase):
def test_get_by_id_create_update_delete(self):
obj = self._test_class(self.context, **self.db_obj)
obj.create()
new = self._test_class.get_by_id(self.context, id=obj.id)
self.assertEqual(obj, new)
obj = new
for key, val in self.db_objs[1].items():
if key not in self._test_class.fields_no_update:
setattr(obj, key, val)
obj.update()
new = self._test_class.get_by_id(self.context, id=obj.id)
self.assertEqual(obj, new)
obj = new
new.delete()
new = self._test_class.get_by_id(self.context, id=obj.id)
self.assertIsNone(new)

0
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/extension_drivers/__init__.py Normal file

88
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/extension_drivers/test_qos_driver.py Normal file

@ -0,0 +1,88 @@
# 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.extensions import qos
from neutron.plugins.ml2.drivers.openvswitch.agent.extension_drivers import (
qos_driver)
from neutron.tests.unit.plugins.ml2.drivers.openvswitch.agent import (
ovs_test_base)
class OVSQoSAgentDriverBwLimitRule(ovs_test_base.OVSAgentConfigTestBase):
def setUp(self):
super(OVSQoSAgentDriverBwLimitRule, self).setUp()
self.qos_driver = qos_driver.QosOVSAgentDriver()
self.qos_driver.initialize()
self.qos_driver.br_int = mock.Mock()
self.qos_driver.br_int.get_qos_bw_limit_for_port = mock.Mock(
return_value=(1000, 10))
self.get = self.qos_driver.br_int.get_qos_bw_limit_for_port
self.qos_driver.br_int.del_qos_bw_limit_for_port = mock.Mock()
self.delete = self.qos_driver.br_int.del_qos_bw_limit_for_port
self.qos_driver.br_int.create_qos_bw_limit_for_port = mock.Mock()
self.create = self.qos_driver.br_int.create_qos_bw_limit_for_port
self.rule = self._create_bw_limit_rule()
self.port = self._create_fake_port()
def _create_bw_limit_rule(self):
return {'type': qos.RULE_TYPE_BANDWIDTH_LIMIT,
'max_kbps': '200',
'max_burst_kbps': '2'}
def _create_fake_port(self):
return {'name': 'fakeport'}
def test_create_new_rule(self):
self.qos_driver.br_int.get_qos_bw_limit_for_port = mock.Mock(
return_value=(None, None))
self.qos_driver.create(self.port, [self.rule])
# Assert create is the last call
self.assertEqual(
'create_qos_bw_limit_for_port',
self.qos_driver.br_int.method_calls[-1][0])
self.assertEqual(0, self.delete.call_count)
self.create.assert_called_once_with(
self.port['name'], self.rule['max_kbps'],
self.rule['max_burst_kbps'])
def test_create_existing_rules(self):
self.qos_driver.create(self.port, [self.rule])
self._assert_rule_create_updated()
def test_update_rules(self):
self.qos_driver.update(self.port, [self.rule])
self._assert_rule_create_updated()
def test_delete_rules(self):
self.qos_driver.delete(self.port, [self.rule])
self.delete.assert_called_once_with(self.port['name'])
def test_unknown_rule_id(self):
self.rule['type'] = 'unknown'
self.qos_driver.create(self.port, [self.rule])
self.assertEqual(0, self.create.call_count)
self.assertEqual(0, self.delete.call_count)
def _assert_rule_create_updated(self):
# Assert create is the last call
self.assertEqual(
'create_qos_bw_limit_for_port',
self.qos_driver.br_int.method_calls[-1][0])
self.delete.assert_called_once_with(self.port['name'])
self.create.assert_called_once_with(
self.port['name'], self.rule['max_kbps'],
self.rule['max_burst_kbps'])

7
neutron/tests/unit/plugins/ml2/drivers/openvswitch/agent/test_ovs_neutron_agent.py

@ -374,7 +374,12 @@ class TestOvsNeutronAgent(object):
return_value=None):
self.assertFalse(get_dev_fn.called)
def test_treat_devices_added_updated_updates_known_port(self):
#TODO(QoS) that this mock should go away once we don't hardcode
#qos extension.
@mock.patch('neutron.api.rpc.handlers.resources_rpc.'
'ResourcesServerRpcApi.get_info', return_value=[])
def test_treat_devices_added_updated_updates_known_port(
self, *args):
details = mock.MagicMock()
details.__contains__.side_effect = lambda x: True
self.assertTrue(self._mock_treat_devices_added_updated(

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

@ -1630,3 +1630,75 @@ class TestMl2PluginCreateUpdateDeletePort(base.BaseTestCase):
# run the transaction balancing function defined in this test
plugin.delete_port(self.context, 'fake_id')
self.assertTrue(self.notify.call_count)
class TestMl2PluginCreateUpdateNetwork(base.BaseTestCase):
def setUp(self):
super(TestMl2PluginCreateUpdateNetwork, self).setUp()
self.context = mock.MagicMock()
self.notify_p = mock.patch('neutron.callbacks.registry.notify')
self.notify = self.notify_p.start()
def _ensure_transaction_is_closed(self):
transaction = self.context.session.begin(subtransactions=True)
enter = transaction.__enter__.call_count
exit = transaction.__exit__.call_count
self.assertEqual(enter, exit)
def _create_plugin_for_create_update_network(self):
plugin = ml2_plugin.Ml2Plugin()
plugin.extension_manager = mock.Mock()
plugin.type_manager = mock.Mock()
plugin.mechanism_manager = mock.Mock()
plugin.notifier = mock.Mock()
mock.patch('neutron.extensions.providernet.'
'_raise_if_updates_provider_attributes').start()
self.notify.side_effect = (
lambda r, e, t, **kwargs: self._ensure_transaction_is_closed())
return plugin
def test_create_network_rpc_outside_transaction(self):
with mock.patch.object(ml2_plugin.Ml2Plugin, '__init__') as init,\
mock.patch.object(base_plugin.NeutronDbPluginV2,
'create_network'):
init.return_value = None
plugin = self._create_plugin_for_create_update_network()
plugin.create_network(self.context, mock.MagicMock())
kwargs = {'context': self.context, 'network': mock.ANY}
self.notify.assert_called_once_with('network', 'after_create',
plugin, **kwargs)
def test_create_network_bulk_rpc_outside_transaction(self):
with mock.patch.object(ml2_plugin.Ml2Plugin, '__init__') as init,\
mock.patch.object(base_plugin.NeutronDbPluginV2,
'create_network'):
init.return_value = None
plugin = self._create_plugin_for_create_update_network()
plugin.create_network_bulk(self.context,
{'networks':
[mock.MagicMock(), mock.MagicMock()]})
self.assertEqual(2, self.notify.call_count)
def test_update_network_rpc_outside_transaction(self):
with mock.patch.object(ml2_plugin.Ml2Plugin, '__init__') as init,\
mock.patch.object(base_plugin.NeutronDbPluginV2,
'update_network'):
init.return_value = None
plugin = self._create_plugin_for_create_update_network()
plugin.update_network(self.context, 'fake_id', mock.MagicMock())
kwargs = {
'context': self.context,
'network': mock.ANY,
}
self.notify.assert_called_once_with('network', 'after_update',
plugin, **kwargs)

0
neutron/tests/unit/services/qos/__init__.py Normal file

148
neutron/tests/unit/services/qos/test_qos_extension.py Normal file

@ -0,0 +1,148 @@
# Copyright (c) 2015 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.
import mock
from neutron.extensions import qos
from neutron.plugins.common import constants as plugin_constants
from neutron.services.qos import qos_extension
from neutron.tests import base
def _get_test_dbdata(qos_policy_id):
return {'id': None, 'qos_policy_binding': {'policy_id': qos_policy_id,
'network_id': 'fake_net_id'}}
class QosResourceExtensionHandlerTestCase(base.BaseTestCase):
def setUp(self):
super(QosResourceExtensionHandlerTestCase, self).setUp()
self.ext_handler = qos_extension.QosResourceExtensionHandler()
policy_p = mock.patch('neutron.objects.qos.policy.QosPolicy')
self.policy_m = policy_p.start()
def test_process_resource_no_qos_policy_id(self):
self.ext_handler.process_resource(None, qos_extension.PORT, {}, None)
self.assertFalse(self.policy_m.called)
def _mock_plugin_loaded(self, plugin_loaded):
plugins = {}
if plugin_loaded:
plugins[plugin_constants.QOS] = None
return mock.patch('neutron.manager.NeutronManager.get_service_plugins',
return_value=plugins)
def test_process_resource_no_qos_plugin_loaded(self):
with self._mock_plugin_loaded(False):
self.ext_handler.process_resource(None, qos_extension.PORT,
{qos.QOS_POLICY_ID: None}, None)
self.assertFalse(self.policy_m.called)
def test_process_resource_port_new_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
actual_port = {'id': mock.Mock(),
qos.QOS_POLICY_ID: qos_policy_id}
qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=qos_policy)
self.ext_handler.process_resource(
None, qos_extension.PORT, {qos.QOS_POLICY_ID: qos_policy_id},
actual_port)
qos_policy.attach_port.assert_called_once_with(actual_port['id'])
def test_process_resource_port_updated_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
port_id = mock.Mock()
actual_port = {'id': port_id,
qos.QOS_POLICY_ID: qos_policy_id}
old_qos_policy = mock.MagicMock()
self.policy_m.get_port_policy = mock.Mock(
return_value=old_qos_policy)
new_qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=new_qos_policy)
self.ext_handler.process_resource(
None, qos_extension.PORT, {qos.QOS_POLICY_ID: qos_policy_id},
actual_port)
old_qos_policy.detach_port.assert_called_once_with(port_id)
new_qos_policy.attach_port.assert_called_once_with(port_id)
def test_process_resource_network_new_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
actual_network = {'id': mock.Mock(),
qos.QOS_POLICY_ID: qos_policy_id}
qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=qos_policy)
self.ext_handler.process_resource(
None, qos_extension.NETWORK,
{qos.QOS_POLICY_ID: qos_policy_id}, actual_network)
qos_policy.attach_network.assert_called_once_with(
actual_network['id'])
def test_process_resource_network_updated_policy(self):
with self._mock_plugin_loaded(True):
qos_policy_id = mock.Mock()
network_id = mock.Mock()
actual_network = {'id': network_id,
qos.QOS_POLICY_ID: qos_policy_id}
old_qos_policy = mock.MagicMock()
self.policy_m.get_network_policy = mock.Mock(
return_value=old_qos_policy)
new_qos_policy = mock.MagicMock()
self.policy_m.get_by_id = mock.Mock(return_value=new_qos_policy)
self.ext_handler.process_resource(
None, qos_extension.NETWORK,
{qos.QOS_POLICY_ID: qos_policy_id}, actual_network)
old_qos_policy.detach_network.assert_called_once_with(network_id)
new_qos_policy.attach_network.assert_called_once_with(network_id)
def test_extract_resource_fields_plugin_not_loaded(self):
with self._mock_plugin_loaded(False):
fields = self.ext_handler.extract_resource_fields(None, None)
self.assertEqual({}, fields)
def _test_extract_resource_fields_for_port(self, qos_policy_id):
with self._mock_plugin_loaded(True):
fields = self.ext_handler.extract_resource_fields(
qos_extension.PORT, _get_test_dbdata(qos_policy_id))
self.assertEqual({qos.QOS_POLICY_ID: qos_policy_id}, fields)
def test_extract_resource_fields_no_port_policy(self):
self._test_extract_resource_fields_for_port(None)
def test_extract_resource_fields_port_policy_exists(self):
qos_policy_id = mock.Mock()
self._test_extract_resource_fields_for_port(qos_policy_id)
def _test_extract_resource_fields_for_network(self, qos_policy_id):
with self._mock_plugin_loaded(True):
fields = self.ext_handler.extract_resource_fields(
qos_extension.NETWORK, _get_test_dbdata(qos_policy_id))
self.assertEqual({qos.QOS_POLICY_ID: qos_policy_id}, fields)
def test_extract_resource_fields_no_network_policy(self):
self._test_extract_resource_fields_for_network(None)
def test_extract_resource_fields_network_policy_exists(self):
qos_policy_id = mock.Mock()
qos_policy = mock.Mock()
qos_policy.id = qos_policy_id
self._test_extract_resource_fields_for_network(qos_policy_id)

1
requirements.txt

@ -36,6 +36,7 @@ oslo.rootwrap>=2.0.0 # Apache-2.0
oslo.serialization>=1.4.0 # Apache-2.0
oslo.service>=0.1.0 # Apache-2.0
oslo.utils>=1.9.0 # Apache-2.0
oslo.versionedobjects>=0.3.0,!=0.5.0
python-novaclient>=2.22.0

6
setup.cfg

@ -141,6 +141,7 @@ neutron.service_plugins =
neutron.services.loadbalancer.plugin.LoadBalancerPlugin = neutron_lbaas.services.loadbalancer.plugin:LoadBalancerPlugin
neutron.services.vpn.plugin.VPNDriverPlugin = neutron_vpnaas.services.vpn.plugin:VPNDriverPlugin
ibm_l3 = neutron.services.l3_router.l3_sdnve:SdnveL3ServicePlugin
qos = neutron.services.qos.qos_plugin:QoSPlugin
neutron.service_providers =
# These are for backwards compat with Juno firewall service provider configuration values
neutron.services.firewall.drivers.linux.iptables_fwaas.IptablesFwaasDriver = neutron_fwaas.services.firewall.drivers.linux.iptables_fwaas:IptablesFwaasDriver
@ -194,11 +195,16 @@ neutron.ml2.extension_drivers =
testdb = neutron.tests.unit.plugins.ml2.drivers.ext_test:TestDBExtensionDriver
port_security = neutron.plugins.ml2.extensions.port_security:PortSecurityExtensionDriver
cisco_n1kv_ext = neutron.plugins.ml2.drivers.cisco.n1kv.n1kv_ext_driver:CiscoN1kvExtensionDriver
qos = neutron.plugins.ml2.extensions.qos:QosExtensionDriver
neutron.openstack.common.cache.backends =
memory = neutron.openstack.common.cache._backends.memory:MemoryBackend
neutron.ipam_drivers =
fake = neutron.tests.unit.ipam.fake_driver:FakeDriver
internal = neutron.ipam.drivers.neutrondb_ipam.driver:NeutronDbPool
neutron.agent.l2.extensions =
qos = neutron.agent.l2.extensions.qos_agent:QosAgentExtension
neutron.qos.agent_drivers =
ovs = neutron.plugins.ml2.drivers.openvswitch.agent.extension_drivers.qos_driver:QosOVSAgentDriver
# These are for backwards compat with Icehouse notification_driver configuration values
oslo.messaging.notify.drivers =
neutron.openstack.common.notifier.log_notifier = oslo_messaging.notify._impl_log:LogDriver