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

Note to reviewers: gerrit diff for merge patches is very limited, and
leaving comments in global section won't scale, so please comment here:

https://etherpad.openstack.org/p/qos-merge-back-review

This merge commit introduces QoS feature into Liberty release of
Neutron.

The feature is documented in: doc/source/devref/quality_of_service.rst
included with the merge patch.

It includes:

- QoS API service plugin with QoS policy and QoS bandwidth limit
  (egress) rule support;
- core plugin mechanism to determine supported rule types, with its ML2
  implementation;
- new agent extension manager;
- QoS agent extension with pluggable backend QoS drivers (Open vSwitch
  and SR-IOV support is included).

To extend network and port core resources with qos_policy_id attribute,
a new ML2 extension driver (qos) was introduced that relies on the QoS
core resource extension (the idea is that eventually we'll get a core
resource extension manager that can be directly reused by core plugins).

Agent-server interaction is based on:

- get_device_details() method that is extended with qos_policy_id;
- a new push/pull mechanism that allows agents and servers to
  communicate using oslo.versionedobjects based objects sent on the
  wire.

The merge includes the following types of test coverage:

- unit tests;
- functional tests for OVS agent, QoS agent extension, and low level
  ovs_lib changes;
- API tests to cover port/network qos_policy_id attribute and new QoS
  resources.

The client changes can be found at:

* https://review.openstack.org/189655
* https://review.openstack.org/198277

The team also prepared fullstack test but it needs to wait for client
merge before it can pass in the gate:

* https://review.openstack.org/202492

Gerrit does not show diff for merge changes that did not result in any
conflict, so to facilitate review, rely on the following steps:

- fetch the patch locally
- git fetch origin
- git diff origin/master...

This merge also disables qos extension API tests until the service is
enabled in master gate.

Local changes apart from conflicts:
- updated down_revision for qos migration to reflect master expand head;
- disabled qos API tests with gate_hook.sh until we have it enabled in
  master gate;
- bumped oslo.versionedobjects requirement to reflect what is in
  openstack/requirements' global-requirements.txt

DocImpact
APIImpact
Partially-Implements: blueprint quantum-qos-api
Partially-Implements: blueprint ml2-qos
Partially-Implements: blueprint ml2-qos-ovs-bwlimiting
Partially-Implements: blueprint ml2-sriov-qos-with-bwlimiting
Change-Id: I92916d0e391791187e9a25ff172fb4b3504857b1
This commit is contained in:
Ihar Hrachyshka 2015-08-17 13:05:32 +02:00
commit 70727ba781
128 changed files with 7120 additions and 333 deletions

View File

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

View File

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

View File

@ -0,0 +1,357 @@
==================
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 extends core resources (ports, networks) without using mixins inherited
from plugins but through an ml2 extension driver.
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. Note that rules are subattributes
of policies and hence embedded into their URIs.
* neutron.services.qos.qos_plugin:
QoSPlugin, service plugin that implements 'qos' extension, receiving and
handling API calls to create/modify policies and rules.
* neutron.services.qos.notification_drivers.manager:
the manager that passes object notifications down to every enabled
notification driver.
* neutron.services.qos.notification_drivers.qos_base:
the interface class for pluggable notification drivers that are used to
update backends about new {create, update, delete} events on any rule or
policy change.
* neutron.services.qos.notification_drivers.message_queue:
MQ-based reference notification driver which updates agents via messaging
bus, using `RPC callbacks <rpc_callbacks.html>`_.
* neutron.core_extensions.base:
Contains an interface class to implement core resource (port/network)
extensions. Core resource extensions are then easily integrated into
interested plugins. We may need to have a core resource extension manager
that would utilize those extensions, to avoid plugin modifications for every
new core resource extension.
* neutron.core_extensions.qos:
Contains QoS core resource extension that conforms to the interface described
above.
* neutron.plugins.ml2.extensions.qos:
Contains ml2 extension driver that handles core resource updates by reusing
the core_extensions.qos module mentioned above. In the future, we would like
to see a plugin-agnostic core resource extension manager that could be
integrated into other plugins with ease.
Supported QoS rule types
------------------------
Any plugin or Ml2 mechanism driver can claim support for some QoS rule types by
providing a plugin/driver class property called 'supported_qos_rule_types' that
should return a list of strings that correspond to QoS rule types (for the list
of all rule types, see: neutron.extensions.qos.VALID_RULE_TYPES).
In the most simple case, the property can be represented by a simple Python
list defined on the class.
For Ml2 plugin, the list of supported QoS rule types is defined as a common
subset of rules supported by all active mechanism drivers.
Note: the list of supported rule types reported by core plugin is not enforced
when accessing QoS rule resources. This is mostly because then we would not be
able to create any rules while at least one ml2 driver in gate lacks support
for QoS (at the moment of writing, linuxbridge is such a driver).
Database models
---------------
QoS design defines the following two conceptual resources to apply 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.
* QosBandwidthLimitRule: defines the only rule type available at the moment.
All database models are defined under:
* neutron.db.qos.models
QoS versioned objects
---------------------
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 was 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 that
belong to the policy. To get a list of all rules for a specific policy, a
consumer of the object can just access the corresponding attribute via:
* policy.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 rules are loaded in a non lazy way, meaning they are all fetched from
the database on policy fetch.
For Qos<type>Rule objects, an extendable approach was taken to allow easy
addition of objects for new rule types. To accomodate this, fields common to
all types are put into 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 suggest 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: those can be reused to fetch other models that do not have
corresponding versioned objects yet, if needed.
* neutron.db.qos.api: contains database functions that are specific to QoS
models.
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.
One more thing to note is that though RPC interface relies on versioned
objects, it does not yet rely on versioning features the oslo.versionedobjects
library provides. This is because Liberty is the first release where we start
using the RPC interface, so we have no way to get different versions in a
cluster. That said, the versioning strategy for QoS is thought through and
described in `the separate page <rpc_callbacks.html>`_.
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.
The flow of updates is as follows:
* if a port that is bound to the agent is attached to a QoS policy, then ML2
plugin detects the change by relying on ML2 QoS extension driver, and
notifies the agent about a port change. The agent proceeds with the
notification by calling to get_device_details() and getting the new port dict
that contains a new qos_policy_id. Each device details dict is passed into l2
agent extension manager that passes it down into every enabled extension,
including QoS. QoS extension sees that there is a new unknown QoS policy for
a port, so it uses ResourcesPullRpcApi to fetch the current state of the
policy (with all the rules included) from the server. After that, the QoS
extension applies the rules by calling into QoS driver that corresponds to
the agent.
* on existing QoS policy update (it includes any policy or its rules change),
server pushes the new policy object state through ResourcesPushRpcApi
interface. The interface fans out the serialized (dehydrated) object to any
agent that is listening for QoS policy updates. If an agent have seen the
policy before (it is attached to one of the ports it maintains), then it goes
with applying the updates to the port. Otherwise, the agent silently ignores
the update.
Agent side design
=================
To ease code reusability between agents and to avoid the need to patch an agent
for each new core resource extension, pluggable L2 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 handle_port events, and do whatever they need
with them.
* neutron.agent.l2.agent_extension:
This module defines an abstract extension interface.
* neutron.agent.l2.extensions.manager:
This module contains a manager that allows to register multiple extensions,
and passes handle_port events down to all enabled extensions.
* neutron.agent.l2.extensions.qos
defines QoS L2 agent extension. It receives handle_port and delete_port
events and passes them down into QoS agent backend driver (see below). The
file also defines the QosAgentDriver interface. Note: each backend implements
its own driver. The driver handles low level interaction with the underlying
networking technology, while the QoS extension handles operations that are
common to all agents.
Agent backends
--------------
At the moment, QoS is supported by Open vSwitch and SR-IOV ml2 drivers.
Each agent backend defines a QoS driver that implements the QosAgentDriver
interface:
* Open vSwitch (QosOVSAgentDriver);
* SR-IOV (QosSRIOVAgentDriver).
Open vSwitch
~~~~~~~~~~~~
Open vSwitch implementation relies on the new ovs_lib OVSBridge functions:
* get_egress_bw_limit_for_port
* create_egress_bw_limit_for_port
* delete_egress_bw_limit_for_port
An egress bandwidth limit is effectively configured on the port by setting
the port Interface parameters ingress_policing_rate and
ingress_policing_burst.
That approach is less flexible than linux-htb, Queues and OvS QoS profiles,
which we may explore in the future, but which will need to be used in
combination with openflow rules.
SR-IOV
~~~~~~
SR-IOV bandwidth limit implementation relies on the new pci_lib function:
* set_vf_max_rate
As the name of the function suggests, the limit is applied on a Virtual
Function (VF).
ip link interface has the following limitation for bandwidth limit: it uses
Mbps as units of bandwidth measurement, not kbps, and does not support float
numbers. So in case the limit is set to something less than 1000 kbps, it's set
to 1 Mbps only. If the limit is set to something that does not divide to 1000
kbps chunks, then the effective limit is rounded to the nearest integer Mbps
value.
Configuration
=============
To enable the service, the following steps should be followed:
On server side:
* enable qos service in service_plugins;
* set the needed notification_drivers in [qos] section (message_queue is the default);
* for ml2, add 'qos' to extension_drivers in [ml2] section.
On agent side (OVS):
* add 'qos' to extensions in [agent] section.
Testing strategy
================
All the code added or extended as part of the effort got reasonable unit test
coverage.
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).
Functional tests
----------------
Additions to ovs_lib to set bandwidth limits on ports are covered in:
* neutron.tests.functional.agent.test_ovs_lib
API tests
---------
API tests for basic CRUD operations for ports, networks, policies, and rules were added in:
* neutron.tests.api.test_qos

View File

@ -0,0 +1,187 @@
=================================
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 and subscribe for resource events.
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:
* QoS policies;
* Security Groups.
Using a remote publisher/subscriber pattern, the information about such
resources could be published using fanout messages 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:
The strategy for upgrades will be:
During upgrades, we pin neutron-server to a compatible version for resource
fanout updates, and the server sends both the old, and the newer version.
The new agents process updates, taking the newer version of the resource
fanout updates. When the whole system upgraded, we un-pin the compatible
version fanout.
Serialized versioned objects look like::
{'versioned_object.version': '1.0',
'versioned_object.name': 'QoSPolicy',
'versioned_object.data': {'rules': [
{'versioned_object.version': '1.0',
'versioned_object.name': 'QoSBandwidthLimitRule',
'versioned_object.data': {'name': u'a'},
'versioned_object.namespace': 'versionedobjects'}
],
'uuid': u'abcde',
'name': u'aaa'},
'versioned_object.namespace': 'versionedobjects'}
Topic names for every resource type RPC endpoint
================================================
neutron-vo-<resource_class_name>-<version>
In the future, we may want to get oslo messaging to support subscribing
topics dynamically, then we may want to use:
neutron-vo-<resource_class_name>-<resource_id>-<version> instead,
or something equivalent which would allow fine granularity for the receivers
to only get interesting information to them.
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.api.rpc.callbacks.consumer import registry
from neutron.api.rpc.callbacks import events
from neutron.api.rpc.callbacks import resources
def process_resource_updates(resource_type, resource, event_type):
# send to the right handler which will update any control plane
# details related to the updated resource...
def subscribe_resources():
registry.subscribe(process_resource_updates, resources.SEC_GROUP)
registry.subscribe(process_resource_updates, resources.QOS_POLICY)
def port_update(port):
# here we extract sg_id and qos_policy_id from port..
sec_group = registry.pull(resources.SEC_GROUP, sg_id)
qos_policy = registry.pull(resources.QOS_POLICY, qos_policy_id)
The relevant function is:
* subscribe(callback, resource_type): subscribes callback to a resource type.
The callback function will receive the following arguments:
* resource_type: the type of resource which is receiving the update.
* resource: resource of supported object
* event_type: will be one of CREATED, UPDATED, or DELETED, see
neutron.api.rpc.callbacks.events for details.
With the underlaying oslo_messaging support for dynamic topics on the receiver
we cannot implement a per "resource type + resource id" topic, rabbitmq seems
to handle 10000's of topics without suffering, but creating 100's of
oslo_messaging receivers on different topics seems to crash.
We may want to look into that later, to avoid agents receiving resource updates
which are uninteresting to them.
Unsubscribing from resources
============================
To unsubscribe registered callbacks:
* unsubscribe(callback, resource_type): unsubscribe from specific resource type.
* unsubscribe_all(): unsubscribe from all resources.
Sending resource events
=======================
On the server side, resource updates could come from anywhere, a service plugin,
an extension, anything that updates, creates, or destroys the resource and that
is of any interest to subscribed agents.
The server/publisher side may look like::
from neutron.api.rpc.callbacks.producer import registry
from neutron.api.rpc.callbacks import events
def create_qos_policy(...):
policy = fetch_policy(...)
update_the_db(...)
registry.push(policy, events.CREATED)
def update_qos_policy(...):
policy = fetch_policy(...)
update_the_db(...)
registry.push(policy, events.UPDATED)
def delete_qos_policy(...):
policy = fetch_policy(...)
update_the_db(...)
registry.push(policy, events.DELETED)
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

View File

@ -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
@ -1028,3 +1028,7 @@ lock_path = $state_path/lock
# Deprecated, use rpc_backend=kombu+memory or rpc_backend=fake (boolean value)
# Deprecated group/name - [DEFAULT]/fake_rabbit
# fake_rabbit = false
[qos]
# Drivers list to use to send the update notification
# notification_drivers = message_queue

View File

@ -133,6 +133,11 @@
#
# quitting_rpc_timeout = 10
# (ListOpt) Extensions list to use
# Example: extensions = qos
#
# extensions =
[securitygroup]
# Firewall driver for realizing neutron security group function.
# firewall_driver = neutron.agent.firewall.NoopFirewallDriver

View File

@ -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",

View File

@ -489,6 +489,36 @@ class OVSBridge(BaseOVS):
txn.add(self.ovsdb.db_set('Controller',
controller_uuid, *attr))
def _set_egress_bw_limit_for_port(self, port_name, max_kbps,
max_burst_kbps):
with self.ovsdb.transaction(check_error=True) as txn:
txn.add(self.ovsdb.db_set('Interface', port_name,
('ingress_policing_rate', max_kbps)))
txn.add(self.ovsdb.db_set('Interface', port_name,
('ingress_policing_burst',
max_burst_kbps)))
def create_egress_bw_limit_for_port(self, port_name, max_kbps,
max_burst_kbps):
self._set_egress_bw_limit_for_port(
port_name, max_kbps, max_burst_kbps)
def get_egress_bw_limit_for_port(self, port_name):
max_kbps = self.db_get_val('Interface', port_name,
'ingress_policing_rate')
max_burst_kbps = self.db_get_val('Interface', port_name,
'ingress_policing_burst')
max_kbps = max_kbps or None
max_burst_kbps = max_burst_kbps or None
return max_kbps, max_burst_kbps
def delete_egress_bw_limit_for_port(self, port_name):
self._set_egress_bw_limit_for_port(
port_name, 0, 0)
def __enter__(self):
self.create()
return self

View File

View 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 extensions.
An agent extension extends the agent core functionality.
"""
def initialize(self, connection, driver_type):
"""Perform agent core resource extension initialization.
:param connection: RPC connection that can be reused by the extension
to define its RPC endpoints
:param driver_type: a string that defines the agent type to the
extension. Can be used to choose the right backend
implementation.
Called after all extensions have been loaded.
No port handling will be called before this method.
"""
@abc.abstractmethod
def handle_port(self, context, data):
"""Handle agent extension for port.
This can be called on either create or update, depending on the
code flow. Thus, it's this function's responsibility to check what
actually changed.
:param context - rpc context
:param data - port data
"""
@abc.abstractmethod
def delete_port(self, context, data):
"""Delete port from agent extension.
:param context - rpc context
:param data - port data
"""

View File

View File

@ -0,0 +1,85 @@
# 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_config import cfg
from oslo_log import log
import stevedore
from neutron.i18n import _LE, _LI
LOG = log.getLogger(__name__)
L2_AGENT_EXT_MANAGER_NAMESPACE = 'neutron.agent.l2.extensions'
L2_AGENT_EXT_MANAGER_OPTS = [
cfg.ListOpt('extensions',
default=[],
help=_('Extensions list to use')),
]
def register_opts(conf):
conf.register_opts(L2_AGENT_EXT_MANAGER_OPTS, 'agent')
class AgentExtensionsManager(stevedore.named.NamedExtensionManager):
"""Manage agent extensions."""
def __init__(self, conf):
super(AgentExtensionsManager, self).__init__(
L2_AGENT_EXT_MANAGER_NAMESPACE, conf.agent.extensions,
invoke_on_load=True, name_order=True)
LOG.info(_LI("Loaded agent extensions: %s"), self.names())
def initialize(self, connection, driver_type):
"""Initialize enabled L2 agent extensions.
:param connection: RPC connection that can be reused by extensions to
define their RPC endpoints
:param driver_type: a string that defines the agent type to the
extension. Can be used by the extension to choose
the right backend implementation.
"""
# Initialize each agent extension in the list.
for extension in self:
LOG.info(_LI("Initializing agent extension '%s'"), extension.name)
extension.obj.initialize(connection, driver_type)
def handle_port(self, context, data):
"""Notify all agent extensions to handle port."""
for extension in self:
try:
extension.obj.handle_port(context, data)
# TODO(QoS) add agent extensions exception and catch them here
except AttributeError:
LOG.exception(
_LE("Agent Extension '%(name)s' failed "
"while handling port update"),
{'name': extension.name}
)
def delete_port(self, context, data):
"""Notify all agent extensions to delete port."""
for extension in self:
try:
extension.obj.delete_port(context, data)
# TODO(QoS) add agent extensions exception and catch them here
# instead of AttributeError
except AttributeError:
LOG.exception(
_LE("Agent Extension '%(name)s' failed "
"while handling port deletion"),
{'name': extension.name}
)

View File

@ -0,0 +1,149 @@
# 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_concurrency import lockutils
import six
from neutron.agent.l2 import agent_extension
from neutron.api.rpc.callbacks.consumer import registry
from neutron.api.rpc.callbacks import events
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):
"""Defines 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.
"""
@abc.abstractmethod
def create(self, port, qos_policy):
"""Apply QoS rules on port for the first time.
:param port: port object.
:param qos_policy: the QoS policy to be applied on port.
"""
#TODO(QoS) we may want to provide default implementations of calling
#delete and then update
@abc.abstractmethod
def update(self, port, qos_policy):
"""Apply QoS rules on port.
:param port: port object.
:param qos_policy: the QoS policy to be applied on port.
"""
@abc.abstractmethod
def delete(self, port, qos_policy):
"""Remove QoS rules from port.
:param port: port object.
:param qos_policy: the QoS policy to be removed from port.
"""
class QosAgentExtension(agent_extension.AgentCoreResourceExtension):
SUPPORTED_RESOURCES = [resources.QOS_POLICY]
def initialize(self, connection, driver_type):
"""Perform Agent Extension initialization.
"""
self.resource_rpc = resources_rpc.ResourcesPullRpcApi()
self.qos_driver = manager.NeutronManager.load_class_for_provider(
'neutron.qos.agent_drivers', driver_type)()
self.qos_driver.initialize()
# we cannot use a dict of sets here because port dicts are not hashable
self.qos_policy_ports = collections.defaultdict(dict)
self.known_ports = set()
registry.subscribe(self._handle_notification, resources.QOS_POLICY)
self._register_rpc_consumers(connection)
def _register_rpc_consumers(self, connection):
endpoints = [resources_rpc.ResourcesPushRpcCallback()]
for resource_type in self.SUPPORTED_RESOURCES:
# we assume that neutron-server always broadcasts the latest
# version known to the agent
topic = resources_rpc.resource_type_versioned_topic(resource_type)
connection.create_consumer(topic, endpoints, fanout=True)
@lockutils.synchronized('qos-port')
def _handle_notification(self, qos_policy, event_type):
# server does not allow to remove a policy that is attached to any
# port, so we ignore DELETED events. Also, if we receive a CREATED
# event for a policy, it means that there are no ports so far that are
# attached to it. That's why we are interested in UPDATED events only
if event_type == events.UPDATED:
self._process_update_policy(qos_policy)
@lockutils.synchronized('qos-port')
def handle_port(self, context, port):
"""Handle agent QoS extension for port.
This method applies a new policy to a port using the QoS driver.
Update events are handled in _handle_notification.
"""
port_id = port['port_id']
qos_policy_id = port.get('qos_policy_id')
if qos_policy_id is None:
self._process_reset_port(port)
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)
qos_policy = self.resource_rpc.pull(
context, resources.QOS_POLICY, qos_policy_id)
self.qos_driver.create(port, qos_policy)
def delete_port(self, context, port):
self._process_reset_port(port)
def _process_update_policy(self, qos_policy):
for port_id, port in self.qos_policy_ports[qos_policy.id].items():
# TODO(QoS): for now, just reflush the rules on the port. Later, we
# may want to apply the difference between the rules lists only.
self.qos_driver.delete(port, None)
self.qos_driver.update(port, qos_policy)
def _process_reset_port(self, port):
port_id = port['port_id']
if port_id in self.known_ports:
self.known_ports.remove(port_id)
for qos_policy_id, port_dict in self.qos_policy_ports.items():
if port_id in port_dict:
del port_dict[port_id]
self.qos_driver.delete(port, None)
return

View File

@ -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

View File

@ -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)

View File

@ -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

View File

@ -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)

View File

View 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 oslo_log import log as logging
from neutron.api.rpc.callbacks import resource_manager
LOG = logging.getLogger(__name__)
#TODO(ajo): consider adding locking to _get_manager, it's
# safe for eventlet, but not for normal threading.
def _get_manager():
return resource_manager.ConsumerResourceCallbacksManager()
def subscribe(callback, resource_type):
_get_manager().register(callback, resource_type)
def unsubscribe(callback, resource_type):
_get_manager().unregister(callback, resource_type)
def push(resource_type, resource, event_type):
"""Push resource events into all registered callbacks for the type."""
callbacks = _get_manager().get_callbacks(resource_type)
for callback in callbacks:
callback(resource, event_type)
def clear():
_get_manager().clear()

View File

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

View File

@ -0,0 +1,25 @@
# 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.common import exceptions
class CallbackWrongResourceType(exceptions.NeutronException):
message = _('Callback for %(resource_type)s returned wrong resource type')
class CallbackNotFound(exceptions.NeutronException):
message = _('Callback for %(resource_type)s not found')
class CallbacksMaxLimitReached(exceptions.NeutronException):
message = _("Cannot add multiple callbacks for %(resource_type)s")

View File

@ -0,0 +1,62 @@
# 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.api.rpc.callbacks import exceptions
from neutron.api.rpc.callbacks import resource_manager
from neutron.objects import base
LOG = logging.getLogger(__name__)
# TODO(ajo): consider adding locking: it's safe for eventlet but not
# for other types of threading.
def _get_manager():
return resource_manager.ProducerResourceCallbacksManager()
def provide(callback, resource_type):
"""Register a callback as a producer for the resource type.
This callback will be used to produce resources of corresponding type for
interested parties.
"""
_get_manager().register(callback, resource_type)
def unprovide(callback, resource_type):
"""Unregister a callback for corresponding resource type."""
_get_manager().unregister(callback, resource_type)
def clear():
"""Clear all callbacks."""
_get_manager().clear()
def pull(resource_type, resource_id, **kwargs):
"""Get resource object that corresponds to resource id.
The function will return an object that is provided by resource producer.
:returns: NeutronObject
"""
callback = _get_manager().get_callback(resource_type)
obj = callback(resource_type, resource_id, **kwargs)
if obj:
if (not isinstance(obj, base.NeutronObject) or
resource_type != obj.obj_name()):
raise exceptions.CallbackWrongResourceType(
resource_type=resource_type)
return obj

View File

@ -0,0 +1,139 @@
# 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_log import log as logging
import six
from neutron.api.rpc.callbacks import exceptions as rpc_exc
from neutron.api.rpc.callbacks import resources
from neutron.callbacks import exceptions
LOG = logging.getLogger(__name__)
# TODO(QoS): split the registry/resources_rpc modules into two separate things:
# one for pull and one for push APIs
def _validate_resource_type(resource_type):
if not resources.is_valid_resource_type(resource_type):
raise exceptions.Invalid(element='resource', value=resource_type)
@six.add_metaclass(abc.ABCMeta)
class ResourceCallbacksManager(object):
"""A callback system that allows information providers in a loose manner.
"""
# This hook is to allow tests to get new objects for the class
_singleton = True
def __new__(cls, *args, **kwargs):
if not cls._singleton:
return super(ResourceCallbacksManager, cls).__new__(cls)
if not hasattr(cls, '_instance'):
cls._instance = super(ResourceCallbacksManager, cls).__new__(cls)
return cls._instance
@abc.abstractmethod
def _add_callback(self, callback, resource_type):
pass
@abc.abstractmethod
def _delete_callback(self, callback, resource_type):
pass
def register(self, callback, resource_type):
"""Register a callback for a resource type.
:param callback: the callback. It must raise or return NeutronObject.
:param resource_type: must be a valid resource type.
"""
LOG.debug("Registering callback for %s", resource_type)
_validate_resource_type(resource_type)
self._add_callback(callback, resource_type)
def unregister(self, callback, resource_type):
"""Unregister callback from the registry.
:param callback: the callback.
:param resource_type: must be a valid resource type.
"""
LOG.debug("Unregistering callback for %s", resource_type)
_validate_resource_type(resource_type)
self._delete_callback(callback, resource_type)
@abc.abstractmethod
def clear(self):
"""Brings the manager to a clean state."""
def get_subscribed_types(self):
return list(self._callbacks.keys())
class ProducerResourceCallbacksManager(ResourceCallbacksManager):
_callbacks = dict()
def _add_callback(self, callback, resource_type):
if resource_type in self._callbacks:
raise rpc_exc.CallbacksMaxLimitReached(resource_type=resource_type)
self._callbacks[resource_type] = callback
def _delete_callback(self, callback, resource_type):
try:
del self._callbacks[resource_type]
except KeyError:
raise rpc_exc.CallbackNotFound(resource_type=resource_type)
def clear(self):
self._callbacks = dict()
def get_callback(self, resource_type):
_validate_resource_type(resource_type)
try:
return self._callbacks[resource_type]
except KeyError:
raise rpc_exc.CallbackNotFound(resource_type=resource_type)
class ConsumerResourceCallbacksManager(ResourceCallbacksManager):
_callbacks = collections.defaultdict(set)
def _add_callback(self, callback, resource_type):
self._callbacks[resource_type].add(callback)
def _delete_callback(self, callback, resource_type):
try:
self._callbacks[resource_type].remove(callback)
if not self._callbacks[resource_type]:
del self._callbacks[resource_type]
except KeyError:
raise rpc_exc.CallbackNotFound(resource_type=resource_type)
def clear(self):
self._callbacks = collections.defaultdict(set)
def get_callbacks(self, resource_type):
"""Return the callback if found, None otherwise.
:param resource_type: must be a valid resource type.
"""
_validate_resource_type(resource_type)
callbacks = self._callbacks[resource_type]
if not callbacks:
raise rpc_exc.CallbackNotFound(resource_type=resource_type)
return callbacks

View File

@ -0,0 +1,49 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from neutron.objects.qos import policy
_QOS_POLICY_CLS = policy.QosPolicy
_VALID_CLS = (
_QOS_POLICY_CLS,
)
_VALID_TYPES = [cls.obj_name() for cls in _VALID_CLS]
# Supported types
QOS_POLICY = _QOS_POLICY_CLS.obj_name()
_TYPE_TO_CLS_MAP = {
QOS_POLICY: _QOS_POLICY_CLS,
}
def get_resource_type(resource_cls):
if not resource_cls:
return None
if not hasattr(resource_cls, 'obj_name'):
return None
return resource_cls.obj_name()
def is_valid_resource_type(resource_type):
return resource_type in _VALID_TYPES
def get_resource_cls(resource_type):
return _TYPE_TO_CLS_MAP.get(resource_type)

View File

@ -0,0 +1,174 @@
# 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.consumer import registry as cons_registry
from neutron.api.rpc.callbacks.producer import registry as prod_registry
from neutron.api.rpc.callbacks import resources
from neutron.common import constants
from neutron.common import exceptions
from neutron.common import rpc as n_rpc
from neutron.common import topics
from neutron.objects import base as obj_base
LOG = logging.getLogger(__name__)
class ResourcesRpcError(exceptions.NeutronException):
pass
class InvalidResourceTypeClass(ResourcesRpcError):
message = _("Invalid resource type %(resource_type)s")
class ResourceNotFound(ResourcesRpcError):
message = _("Resource %(resource_id)s of type %(resource_type)s "
"not found")
def _validate_resource_type(resource_type):
if not resources.is_valid_resource_type(resource_type):
raise InvalidResourceTypeClass(resource_type=resource_type)
def resource_type_versioned_topic(resource_type):
_validate_resource_type(resource_type)
cls = resources.get_resource_cls(resource_type)
return topics.RESOURCE_TOPIC_PATTERN % {'resource_type': resource_type,
'version': cls.VERSION}
class ResourcesPullRpcApi(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: ResourcesPullRpcCallback. For more information on
this RPC interface, see doc/source/devref/rpc_callbacks.rst.
"""
def __new__(cls):
# make it a singleton
if not hasattr(cls, '_instance'):
cls._instance = super(ResourcesPullRpcApi, cls).__new__(cls)
target = oslo_messaging.Target(
topic=topics.PLUGIN, version='1.0',
namespace=constants.RPC_NAMESPACE_RESOURCES)
cls._instance.client = n_rpc.get_client(target)
return cls._instance
@log_helpers.log_method_call
def pull(self, context, resource_type, resource_id):
_validate_resource_type(resource_type)
# we've already validated the resource type, so we are pretty sure the
# class is there => no need to validate it specifically
resource_type_cls = resources.get_resource_cls(resource_type)
cctxt = self.client.prepare()
primitive = cctxt.call(context, 'pull',
resource_type=resource_type,
version=resource_type_cls.VERSION, resource_id=resource_id)
if primitive is None:
raise ResourceNotFound(resource_type=resource_type,
resource_id=resource_id)
return resource_type_cls.clean_obj_from_primitive(primitive)
class ResourcesPullRpcCallback(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: ResourcesPullRpcApi. For more information on
this RPC interface, see doc/source/devref/rpc_callbacks.rst.
"""
# History
# 1.0 Initial version
target = oslo_messaging.Target(
version='1.0', namespace=constants.RPC_NAMESPACE_RESOURCES)
def pull(self, context, resource_type, version, resource_id):
obj = prod_registry.pull(resource_type, resource_id, context=context)
if obj:
#TODO(QoS): Remove in the future with new version of
# versionedobjects containing
# https://review.openstack.org/#/c/207998/
if version == obj.VERSION:
version = None
return obj.obj_to_primitive(target_version=version)
class ResourcesPushRpcApi(object):
"""Plugin-side RPC for plugin-to-agents interaction.
This interface is designed to push versioned object updates to interested
agents using fanout topics.
This class implements the caller side of an rpc interface. The receiver
side can be found below: ResourcesPushRpcCallback.
"""
def __init__(self):
target = oslo_messaging.Target(
version='1.0',
namespace=constants.RPC_NAMESPACE_RESOURCES)
self.client = n_rpc.get_client(target)
def _prepare_object_fanout_context(self, obj):
"""Prepare fanout context, one topic per object type."""
obj_topic = resource_type_versioned_topic(obj.obj_name())
return self.client.prepare(fanout=True, topic=obj_topic)
@log_helpers.log_method_call
def push(self, context, resource, event_type):
resource_type = resources.get_resource_type(resource)
_validate_resource_type(resource_type)
cctxt = self._prepare_object_fanout_context(resource)
#TODO(QoS): Push notifications for every known version once we have
# multiple of those
dehydrated_resource = resource.obj_to_primitive()
cctxt.cast(context, 'push',
resource=dehydrated_resource,
event_type=event_type)
class ResourcesPushRpcCallback(object):
"""Agent-side RPC for plugin-to-agents interaction.
This class implements the receiver for notification about versioned objects
resource updates used by neutron.api.rpc.callbacks. You can find the
caller side in ResourcesPushRpcApi.
"""
# History
# 1.0 Initial version
target = oslo_messaging.Target(version='1.0',
namespace=constants.RPC_NAMESPACE_RESOURCES)
def push(self, context, resource, event_type):
resource_obj = obj_base.NeutronObject.clean_obj_from_primitive(
resource)
LOG.debug("Resources notification (%(event_type)s): %(resource)s",
{'event_type': event_type, 'resource': repr(resource_obj)})
resource_type = resources.get_resource_type(resource_obj)
cons_registry.push(resource_type, resource_obj, event_type)

View File

@ -127,22 +127,24 @@ def arp_header_match_supported():
def vf_management_supported():
is_supported = True
required_caps = (
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_STATE,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_SPOOFCHK)
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_SPOOFCHK,
ip_link_support.IpLinkConstants.IP_LINK_CAPABILITY_RATE)
try:
vf_section = ip_link_support.IpLinkSupport.get_vf_mgmt_section()
for cap in required_caps:
if not ip_link_support.IpLinkSupport.vf_mgmt_capability_supported(
vf_section, cap):
is_supported = False
LOG.debug("ip link command does not support "
"vf capability '%(cap)s'", cap)
return False
except ip_link_support.UnsupportedIpLinkCommand:
LOG.exception(_LE("Unexpected exception while checking supported "
"ip link command"))
return False
return True
return is_supported
def netns_read_requires_helper():

View File

@ -183,6 +183,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

View File

@ -77,6 +77,10 @@ class AdminRequired(NotAuthorized):
message = _("User does not have admin privileges: %(reason)s")
class ObjectNotFound(NotFound):
message = _("Object %(id)s not found.")
class NetworkNotFound(NotFound):
message = _("Network %(net_id)s could not be found")
@ -93,11 +97,30 @@ class PortNotFound(NotFound):
message = _("Port %(port_id)s could not be found")
class QosPolicyNotFound(NotFound):
message = _("QoS policy %(policy_id)s could not be found")
class QosRuleNotFound(NotFound):
message = _("QoS rule %(rule_id)s for policy %(policy_id)s "
"could not be found")
class PortNotFoundOnNetwork(NotFound):
message = _("Port %(port_id)s could not be found "
"on network %(net_id)s")
class PortQosBindingNotFound(NotFound):
message = _("QoS binding for port %(port_id)s and policy %(policy_id)s "
"could not be found")
class NetworkQosBindingNotFound(NotFound):
message = _("QoS binding for network %(net_id)s and policy %(policy_id)s "
"could not be found")
class PolicyFileNotFound(NotFound):
message = _("Policy configuration policy.json could not be found")
@ -118,6 +141,11 @@ class InUse(NeutronException):
message = _("The resource is inuse")
class QosPolicyInUse(InUse):
message = _("QoS Policy %(policy_id)s is used by "
"%(object_type)s %(object_id)s.")
class NetworkInUse(InUse):
message = _("Unable to complete operation on network %(net_id)s. "
"There are one or more ports still in use on the network.")
@ -489,3 +517,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')

View File

@ -19,6 +19,7 @@ PORT = 'port'
SECURITY_GROUP = 'security_group'
L2POPULATION = 'l2population'
DVR = 'dvr'
RESOURCES = 'resources'
CREATE = 'create'
DELETE = 'delete'
@ -37,6 +38,8 @@ DHCP_AGENT = 'dhcp_agent'
METERING_AGENT = 'metering_agent'
LOADBALANCER_AGENT = 'n-lbaas_agent'
RESOURCE_TOPIC_PATTERN = "neutron-vo-%(resource_type)s-%(version)s"
def get_topic_name(prefix, table, operation, host=None):
"""Create a topic name.

View File

@ -19,6 +19,7 @@
"""Utilities and helper functions."""
import datetime
import decimal
import errno
import functools
import hashlib
@ -437,3 +438,14 @@ class DelayedStringRenderer(object):
def __str__(self):
return str(self.function(*self.args, **self.kwargs))
def camelize(s):
return ''.join(s.replace('_', ' ').title().split())
def round_val(val):
# we rely on decimal module since it behaves consistently across Python
# versions (2.x vs. 3.x)
return int(decimal.Decimal(val).quantize(decimal.Decimal('1'),
rounding=decimal.ROUND_HALF_UP))

View File

View File

@ -0,0 +1,48 @@
# 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 six
NETWORK = 'network'
PORT = 'port'
CORE_RESOURCES = [NETWORK, PORT]
@six.add_metaclass(abc.ABCMeta)
class CoreResourceExtension(object):
@abc.abstractmethod
def process_fields(self, context, resource_type,
requested_resource, actual_resource):
"""Process extension fields.
:param context: neutron api request context
:param resource_type: core resource type (one of CORE_RESOURCES)
:param requested_resource: resource dict that contains extension fields
:param actual_resource: actual resource dict known to plugin
"""
@abc.abstractmethod
def extract_fields(self, resource_type, resource):
"""Extract extension fields.
:param resource_type: core resource type (one of CORE_RESOURCES)
:param resource: resource dict that contains extension fields
"""

View 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.common import exceptions as n_exc
from neutron.core_extensions import base
from neutron.db import api as db_api
from neutron import manager
from neutron.objects.qos import policy as policy_object
from neutron.plugins.common import constants as plugin_constants
from neutron.services.qos import qos_consts
class QosCoreResourceExtension(base.CoreResourceExtension):
@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):
obj = policy_object.QosPolicy.get_by_id(context, policy_id)
if obj is None:
raise n_exc.QosPolicyNotFound(policy_id=policy_id)
return obj
def _update_port_policy(self, context, port, port_changes):
old_policy = policy_object.QosPolicy.get_port_policy(
context, port['id'])
if old_policy:
old_policy.detach_port(port['id'])
qos_policy_id = port_changes.get(qos_consts.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_consts.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_consts.QOS_POLICY_ID)
if qos_policy_id is not None:
policy = self._get_policy_obj(context, qos_policy_id)
policy.attach_network(network['id'])
network[qos_consts.QOS_POLICY_ID] = qos_policy_id
def _exec(self, method_name, context, kwargs):
with db_api.autonested_transaction(context.session):
return getattr(self, method_name)(context=context, **kwargs)
def process_fields(self, context, resource_type,
requested_resource, actual_resource):
if (qos_consts.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_fields(self, resource_type, resource):
if not self.plugin_loaded:
return {}
binding = resource['qos_policy_binding']
qos_policy_id = binding['policy_id'] if binding else None
return {qos_consts.QOS_POLICY_ID: qos_policy_id}

View File

@ -20,9 +20,13 @@ 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.common import exceptions as n_exc
from neutron.db import common_db_mixin
_FACADE = None
@ -88,3 +92,48 @@ 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 _safe_get_object(context, model, id):
db_obj = get_object(context, model, id=id)
if db_obj is None:
raise n_exc.ObjectNotFound(id=id)
return db_obj
def update_object(context, model, id, values):
with context.session.begin(subtransactions=True):
db_obj = _safe_get_object(context, model, 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 = _safe_get_object(context, model, id)
context.session.delete(db_obj)

View File

@ -29,6 +29,40 @@ from neutron.db import models_v2
LOG = logging.getLogger(__name__)
def convert_result_to_dict(f):
@functools.wraps(f)
def inner(*args, **kwargs):
result = f(*args, **kwargs)
if result is None:
return None
elif isinstance(result, list):
return [r.to_dict() for r in result]
else:
return result.to_dict()
return inner
def filter_fields(f):
@functools.wraps(f)
def inner_filter(*args, **kwargs):
result = f(*args, **kwargs)
fields = kwargs.get('fields')
if not fields:
try:
pos = f.func_code.co_varnames.index('fields')
fields = args[pos]
except (IndexError, ValueError):
return result
do_filter = lambda d: {k: v for k, v in d.items() if k in fields}
if isinstance(result, list):
return [do_filter(obj) for obj in result]
else:
return do_filter(result)
return inner_filter
class DbBasePluginCommon(common_db_mixin.CommonDbMixin):
"""Stores getters and helper methods for db_base_plugin_v2

View File

@ -1,3 +1,3 @@
1b4c6e320f79
2a16083502f3
48153cb5f051
kilo

View File

@ -0,0 +1,69 @@
# 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: 1b4c6e320f79
Create Date: 2015-06-24 17:03:34.965101
"""
# revision identifiers, used by Alembic.
revision = '48153cb5f051'
down_revision = '1b4c6e320f79'
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(), nullable=False),
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_bandwidth_limit_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, unique=True),
sa.Column('max_kbps', sa.Integer()),
sa.Column('max_burst_kbps', sa.Integer()))

View File

@ -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.quota import models # noqa
from neutron.db import rbac_db_models # noqa
from neutron.db import securitygroups_db # noqa

View File

65
neutron/db/qos/api.py Normal file
View File

@ -0,0 +1,65 @@
# 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_db import exception as oslo_db_exception
from sqlalchemy.orm import exc as orm_exc
from neutron.common import exceptions as n_exc
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):
try:
with context.session.begin(subtransactions=True):
db_obj = models.QosNetworkPolicyBinding(policy_id=policy_id,
network_id=network_id)
context.session.add(db_obj)
except oslo_db_exception.DBReferenceError:
raise n_exc.NetworkQosBindingNotFound(net_id=network_id,
policy_id=policy_id)
def delete_policy_network_binding(context, policy_id, network_id):
try:
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)
except orm_exc.NoResultFound:
raise n_exc.NetworkQosBindingNotFound(net_id=network_id,
policy_id=policy_id)
def create_policy_port_binding(context, policy_id, port_id):
try:
with context.session.begin(subtransactions=True):
db_obj = models.QosPortPolicyBinding(policy_id=policy_id,
port_id=port_id)
context.session.add(db_obj)
except oslo_db_exception.DBReferenceError:
raise n_exc.PortQosBindingNotFound(port_id=port_id,
policy_id=policy_id)
def delete_policy_port_binding(context, policy_id, port_id):
try:
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)
except orm_exc.NoResultFound:
raise n_exc.PortQosBindingNotFound(port_id=port_id,
policy_id=policy_id)

86
neutron/db/qos/models.py Executable file
View File

@ -0,0 +1,86 @@
# 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, nullable=False)
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 QosRuleColumns(models_v2.HasId):
# NOTE(ihrachyshka): we may need to rework it later when we introduce types
# that should not enforce uniqueness
qos_policy_id = sa.Column(sa.String(36), nullable=False, unique=True)
__table_args__ = (
sa.ForeignKeyConstraint(['qos_policy_id'], ['qos_policies.id']),
model_base.BASEV2.__table_args__
)
class QosBandwidthLimitRule(QosRuleColumns, model_base.BASEV2):
__tablename__ = 'qos_bandwidth_limit_rules'
max_kbps = sa.Column(sa.Integer)
max_burst_kbps = sa.Column(sa.Integer)

236
neutron/extensions/qos.py Normal file
View File

@ -0,0 +1,236 @@
# 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.qos import qos_consts
from neutron.services import service_base
QOS_PREFIX = "/qos"
# Attribute Map
QOS_RULE_COMMON_FIELDS = {
'id': {'allow_post': False, 'allow_put': False,
'validate': {'type:uuid': None},
'is_visible': True,
'primary_key': True},
'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},
'rules': {'allow_post': False, 'allow_put': False, '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}}})
}
}
EXTENDED_ATTRIBUTES_2_0 = {
'ports': {qos_consts.QOS_POLICY_ID: {
'allow_post': True,
'allow_put': True,
'is_visible': True,
'default': None,
'validate': {'type:uuid_or_none': None}}},
'networks': {qos_consts.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

View File

156
neutron/objects/base.py Normal file
View File

@ -0,0 +1,156 @@
# 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_db import exception as obj_exc
from oslo_versionedobjects import base as obj_base
import six
from neutron.common import exceptions
from neutron.db import api as db_api
class NeutronObjectUpdateForbidden(exceptions.NeutronException):
message = _("Unable to update the following object fields: %(fields)s")
class NeutronObjectDuplicateEntry(exceptions.Conflict):
message = _("Failed to create a duplicate object")
def get_updatable_fields(cls, fields):
fields = fields.copy()
for field in cls.fields_no_update:
if field in fields:
del fields[field]
return fields
@six.add_metaclass(abc.ABCMeta)
class NeutronObject(obj_base.VersionedObject,
obj_base.VersionedObjectDictCompat,
obj_base.ComparableVersionedObject):
synthetic_fields = []
def __init__(self, context=None, **kwargs):
super(NeutronObject, self).__init__(context, **kwargs)
self.obj_set_defaults()
def to_dict(self):
return dict(self.items())
@classmethod
def clean_obj_from_primitive(cls, primitive, context=None):
obj = cls.obj_from_primitive(primitive, context)
obj.obj_reset_changes()
return obj
@classmethod
def get_by_id(cls, context, id):
raise NotImplementedError()
@classmethod
def validate_filters(cls, **kwargs):
bad_filters = [key for key in kwargs
if key not in cls.fields or key in cls.synthetic_fields]
if bad_filters:
bad_filters = ', '.join(bad_filters)
msg = _("'%s' is not supported for filtering") % bad_filters
raise exceptions.InvalidInput(error_message=msg)
@classmethod
@abc.abstractmethod
def get_objects(cls, context, **kwargs):
raise NotImplementedError()
def create(self):
raise NotImplementedError()
def update(self):
raise NotImplementedError()
def delete(self):
raise NotImplementedError()
class NeutronDbObject(NeutronObject):
# should be overridden for all persistent objects
db_model = None
fields_no_update = []
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()
@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):
cls.validate_filters(**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 _validate_changed_fields(self, fields):
fields = fields.copy()
# We won't allow id update anyway, so let's pop it out not to trigger
# update on id field touched by the consumer
fields.pop('id', None)
forbidden_updates = set(self.fields_no_update) & set(fields.keys())
if forbidden_updates:
raise NeutronObjectUpdateForbidden(fields=forbidden_updates)
return fields
def create(self):
fields = self._get_changed_persistent_fields()
try:
db_obj = db_api.create_object(self._context, self.db_model, fields)
except obj_exc.DBDuplicateEntry:
raise NeutronObjectDuplicateEntry()
self.from_db_object(db_obj)
def update(self):
updates = self._get_changed_persistent_fields()
updates = self._validate_changed_fields(updates)
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)

View File

View File

@ -0,0 +1,163 @@
# 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.
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
from neutron.common import exceptions
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.objects import base
from neutron.objects.qos import rule as rule_obj_impl
@obj_base.VersionedObjectRegistry.register
class QosPolicy(base.NeutronDbObject):
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(default=False),
'rules': obj_fields.ListOfObjectsField('QosRule', subclasses=True),
}
fields_no_update = ['id', 'tenant_id']
synthetic_fields = ['rules']
def to_dict(self):
dict_ = super(QosPolicy, self).to_dict()
if 'rules' in dict_:
dict_['rules'] = [rule.to_dict() for rule in dict_['rules']]
return dict_
def obj_load_attr(self, attrname):
if attrname != 'rules':
raise exceptions.ObjectActionError(
action='obj_load_attr', reason='unable to load %s' % attrname)
if not hasattr(self, attrname):
self.reload_rules()
def reload_rules(self):
rules = rule_obj_impl.get_rules(self._context, self.id)
setattr(self, 'rules', rules)
self.obj_reset_changes(['rules'])
@staticmethod
def _is_policy_accessible(context, db_obj):
#TODO(QoS): Look at I3426b13eede8bfa29729cf3efea3419fb91175c4 for
# other possible solutions to this.
return (context.is_admin or
db_obj.shared or
db_obj.tenant_id == context.tenant_id)
@classmethod
def get_by_id(cls, context, id):
# We want to get the policy regardless of its tenant id. We'll make
# sure the tenant has permission to access the policy later on.
admin_context = context.elevated()
with db_api.autonested_transaction(admin_context.session):
policy_obj = super(QosPolicy, cls).get_by_id(admin_context, id)
if (not policy_obj or
not cls._is_policy_accessible(context, policy_obj)):
return
policy_obj.reload_rules()
return policy_obj
@classmethod
def get_objects(cls, context, **kwargs):
# We want to get the policy regardless of its tenant id. We'll make
# sure the tenant has permission to access the policy later on.
admin_context = context.elevated()
with db_api.autonested_transaction(admin_context.session):
objs = super(QosPolicy, cls).get_objects(admin_context,
**kwargs)
result = []
for obj in objs:
if not cls._is_policy_accessible(context, obj):
continue
obj.reload_rules()
result.append(obj)
return result
@classmethod
def _get_object_policy(cls, context, model, **kwargs):
with db_api.autonested_transaction(context.session):
binding_db_obj = db_api.get_object(context, model, **kwargs)
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)
# TODO(QoS): Consider extending base to trigger registered methods for us
def create(self):
with db_api.autonested_transaction(self._context.session):
super(QosPolicy, self).create()
self.reload_rules()
def delete(self):
models = (
('network', self.network_binding_model),
('port', self.port_binding_model)
)
with db_api.autonested_transaction(self._context.session):
for object_type, model in models:
binding_db_obj = db_api.get_object(self._context, model,
policy_id=self.id)
if binding_db_obj:
raise exceptions.QosPolicyInUse(
policy_id=self.id,
object_type=object_type,
object_id=binding_db_obj['%s_id' % object_type])
super(QosPolicy, self).delete()
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)

View File

@ -0,0 +1,71 @@
# 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
import sys
from oslo_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
import six
from neutron.common import utils
from neutron.db import api as db_api
from neutron.db.qos import models as qos_db_model
from neutron.objects import base
from neutron.services.qos import qos_consts
def get_rules(context, qos_policy_id):
all_rules = []
with db_api.autonested_transaction(context.session):
for rule_type in qos_consts.VALID_RULE_TYPES:
rule_cls_name = 'Qos%sRule' % utils.camelize(rule_type)
rule_cls = getattr(sys.modules[__name__], rule_cls_name)
rules = rule_cls.get_objects(context, qos_policy_id=qos_policy_id)
all_rules.extend(rules)
return all_rules
@six.add_metaclass(abc.ABCMeta)
class QosRule(base.NeutronDbObject):
fields = {
'id': obj_fields.UUIDField(),
'qos_policy_id': obj_fields.UUIDField()
}
fields_no_update = ['id', 'qos_policy_id']
# should be redefined in subclasses
rule_type = None
def to_dict(self):
dict_ = super(QosRule, self).to_dict()
dict_['type'] = self.rule_type
return dict_
@obj_base.VersionedObjectRegistry.register
class QosBandwidthLimitRule(QosRule):
db_model = qos_db_model.QosBandwidthLimitRule
fields = {
'max_kbps': obj_fields.IntegerField(nullable=True),
'max_burst_kbps': obj_fields.IntegerField(nullable=True)
}
rule_type = qos_consts.RULE_TYPE_BANDWIDTH_LIMIT

View File

@ -0,0 +1,42 @@
# 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_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
from neutron import manager
from neutron.objects import base
from neutron.services.qos import qos_consts
class RuleTypeField(obj_fields.BaseEnumField):
def __init__(self, **kwargs):
self.AUTO_TYPE = obj_fields.Enum(
valid_values=qos_consts.VALID_RULE_TYPES)
super(RuleTypeField, self).__init__(**kwargs)
@obj_base.VersionedObjectRegistry.register
class QosRuleType(base.NeutronObject):
fields = {
'type': RuleTypeField(),
}
# we don't receive context because we don't need db access at all
@classmethod
def get_objects(cls, **kwargs):
cls.validate_filters(**kwargs)
core_plugin = manager.NeutronManager.get_plugin()
return [cls(type=type_)
for type_ in core_plugin.supported_qos_rule_types]

View File

@ -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

View File

@ -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

View File

@ -20,6 +20,7 @@ import re
from oslo_log import log as logging
import six
from neutron.common import utils
from neutron.i18n import _LE, _LW
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
@ -144,11 +145,7 @@ class EmbSwitch(object):
@param pci_slot: Virtual Function address
"""
vf_index = self.pci_slot_map.get(pci_slot)
if vf_index is None:
LOG.warning(_LW("Cannot find vf index for pci slot %s"),
pci_slot)
raise exc.InvalidPciSlotError(pci_slot=pci_slot)
vf_index = self._get_vf_index(pci_slot)
return self.pci_dev_wrapper.get_vf_state(vf_index)
def set_device_state(self, pci_slot, state):
@ -157,12 +154,48 @@ class EmbSwitch(object):
@param pci_slot: Virtual Function address
@param state: link state
"""
vf_index = self._get_vf_index(pci_slot)
return self.pci_dev_wrapper.set_vf_state(vf_index, state)
def set_device_max_rate(self, pci_slot, max_kbps):
"""Set device max rate.
@param pci_slot: Virtual Function address
@param max_kbps: device max rate in kbps
"""
vf_index = self._get_vf_index(pci_slot)
#(Note): ip link set max rate in Mbps therefore
#we need to convert the max_kbps to Mbps.
#Zero means to disable the rate so the lowest rate
#available is 1Mbps. Floating numbers are not allowed
if max_kbps > 0 and max_kbps < 1000:
max_mbps = 1
else:
max_mbps = utils.round_val(max_kbps / 1000.0)
log_dict = {
'max_rate': max_mbps,
'max_kbps': max_kbps,
'vf_index': vf_index
}
if max_kbps % 1000 != 0:
LOG.debug("Maximum rate for SR-IOV ports is counted in Mbps; "
"setting %(max_rate)s Mbps limit for port %(vf_index)s "
"instead of %(max_kbps)s kbps",
log_dict)
else:
LOG.debug("Setting %(max_rate)s Mbps limit for port %(vf_index)s",
log_dict)
return self.pci_dev_wrapper.set_vf_max_rate(vf_index, max_mbps)
def _get_vf_index(self, pci_slot):
vf_index = self.pci_slot_map.get(pci_slot)
if vf_index is None:
LOG.warning(_LW("Cannot find vf index for pci slot %s"),
pci_slot)
raise exc.InvalidPciSlotError(pci_slot=pci_slot)
return self.pci_dev_wrapper.set_vf_state(vf_index, state)
return vf_index
def set_device_spoofcheck(self, pci_slot, enabled):
"""Set device spoofchecking
@ -194,16 +227,13 @@ class EmbSwitch(object):
class ESwitchManager(object):
"""Manages logical Embedded Switch entities for physical network."""
def __init__(self, device_mappings, exclude_devices):
"""Constructor.
Create Embedded Switch logical entities for all given device mappings,
using exclude devices.
"""
self.emb_switches_map = {}
self.pci_slot_map = {}
self._discover_devices(device_mappings, exclude_devices)
def __new__(cls):
# make it a singleton
if not hasattr(cls, '_instance'):
cls._instance = super(ESwitchManager, cls).__new__(cls)
cls.emb_switches_map = {}
cls.pci_slot_map = {}
return cls._instance
def device_exists(self, device_mac, pci_slot):
"""Verify if device exists.
@ -250,6 +280,19 @@ class ESwitchManager(object):
return embedded_switch.get_device_state(pci_slot)
return False
def set_device_max_rate(self, device_mac, pci_slot, max_kbps):
"""Set device max rate
Sets the device max rate in kbps
@param device_mac: device mac
@param pci_slot: pci slot
@param max_kbps: device max rate in kbps
"""
embedded_switch = self._get_emb_eswitch(device_mac, pci_slot)
if embedded_switch:
embedded_switch.set_device_max_rate(pci_slot,
max_kbps)
def set_device_state(self, device_mac, pci_slot, admin_state_up):
"""Set device state
@ -276,7 +319,7 @@ class ESwitchManager(object):
embedded_switch.set_device_spoofcheck(pci_slot,
enabled)
def _discover_devices(self, device_mappings, exclude_devices):
def discover_devices(self, device_mappings, exclude_devices):
"""Discover which Virtual functions to manage.
Discover devices, and create embedded switch object for network device
@ -311,3 +354,17 @@ class ESwitchManager(object):
{"device_mac": device_mac, "pci_slot": pci_slot})
embedded_switch = None
return embedded_switch
def get_pci_slot_by_mac(self, device_mac):
"""Get pci slot by mac.
Get pci slot by device mac
@param device_mac: device mac
"""
result = None
for pci_slot, embedded_switch in self.pci_slot_map.items():
used_device_mac = embedded_switch.get_pci_device(pci_slot)
if used_device_mac == device_mac:
result = pci_slot
break
return result

View File

@ -0,0 +1,84 @@
# Copyright 2015 Mellanox Technologies, Ltd
#
# 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.agent.l2.extensions import qos
from neutron.i18n import _LE, _LI, _LW
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import (
exceptions as exc)
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
from neutron.plugins.ml2.drivers.mech_sriov.mech_driver import (
mech_driver)
LOG = logging.getLogger(__name__)
class QosSRIOVAgentDriver(qos.QosAgentDriver):
_SUPPORTED_RULES = (
mech_driver.SriovNicSwitchMechanismDriver.supported_qos_rule_types)
def __init__(self):
super(QosSRIOVAgentDriver, self).__init__()
self.eswitch_mgr = None
def initialize(self):
self.eswitch_mgr = esm.ESwitchManager()
def create(self, port, qos_policy):
self._handle_rules('create', port, qos_policy)
def update(self, port, qos_policy):
self._handle_rules('update', port, qos_policy)
def delete(self, port, qos_policy):
# TODO(QoS): consider optimizing flushing of all QoS rules from the
# port by inspecting qos_policy.rules contents
self._delete_bandwidth_limit(port)
def _handle_rules(self, action, port, qos_policy):
for rule in qos_policy.rules:
if rule.rule_type in self._SUPPORTED_RULES:
handler_name = ("".join(("_", action, "_", rule.rule_type)))
handler = getattr(self, handler_name)
handler(port, rule)
else:
LOG.warning(_LW('Unsupported QoS rule type for %(rule_id)s: '
'%(rule_type)s; skipping'),
{'rule_id': rule.id, 'rule_type': rule.rule_type})
def _create_bandwidth_limit(self, port, rule):
self._update_bandwidth_limit(port, rule)
def _update_bandwidth_limit(self, port, rule):
pci_slot = port['profile'].get('pci_slot')
device = port['device']
self._set_vf_max_rate(device, pci_slot, rule.max_kbps)
def _delete_bandwidth_limit(self, port):
pci_slot = port['profile'].get('pci_slot')
device = port['device']
self._set_vf_max_rate(device, pci_slot)
def _set_vf_max_rate(self, device, pci_slot, max_kbps=0):
if self.eswitch_mgr.device_exists(device, pci_slot):
try:
self.eswitch_mgr.set_device_max_rate(
device, pci_slot, max_kbps)
except exc.SriovNicError:
LOG.exception(
_LE("Failed to set device %s max rate"), device)
else:
LOG.info(_LI("No device with MAC %s defined on agent."), device)

View File

@ -122,6 +122,21 @@ class PciDeviceIPWrapper(ip_lib.IPWrapper):
raise exc.IpCommandError(dev_name=self.dev_name,
reason=str(e))
def set_vf_max_rate(self, vf_index, max_tx_rate):
"""sets vf max rate.
@param vf_index: vf index
@param max_tx_rate: vf max tx rate in Mbps
"""
try:
self._as_root([], "link", ("set", self.dev_name, "vf",
str(vf_index), "rate",
str(max_tx_rate)))
except Exception as e:
LOG.exception(_LE("Failed executing ip command"))
raise exc.IpCommandError(dev_name=self.dev_name,
reason=e)
def _get_vf_link_show(self, vf_list, link_show_out):
"""Get link show output for VFs

View File

@ -26,6 +26,7 @@ from oslo_log import log as logging
import oslo_messaging
from oslo_service import loopingcall
from neutron.agent.l2.extensions import manager as ext_manager
from neutron.agent import rpc as agent_rpc
from neutron.agent import securitygroups_rpc as sg_rpc
from neutron.common import config as common_config
@ -34,7 +35,7 @@ from neutron.common import topics
from neutron.common import utils as n_utils
from neutron import context
from neutron.i18n import _LE, _LI, _LW
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import config # noqa
from neutron.plugins.ml2.drivers.mech_sriov.agent.common import config
from neutron.plugins.ml2.drivers.mech_sriov.agent.common \
import exceptions as exc
from neutron.plugins.ml2.drivers.mech_sriov.agent import eswitch_manager as esm
@ -72,12 +73,13 @@ class SriovNicSwitchAgent(object):
polling_interval):
self.polling_interval = polling_interval
self.conf = cfg.CONF
self.setup_eswitch_mgr(physical_devices_mappings,
exclude_devices)
configurations = {'device_mappings': physical_devices_mappings}
self.agent_state = {
'binary': 'neutron-sriov-nic-agent',
'host': cfg.CONF.host,
'host': self.conf.host,
'topic': n_constants.L2_AGENT_TOPIC,
'configurations': configurations,
'agent_type': n_constants.AGENT_TYPE_NIC_SWITCH,
@ -92,6 +94,10 @@ class SriovNicSwitchAgent(object):
self.sg_agent = sg_rpc.SecurityGroupAgentRpc(self.context,
self.sg_plugin_rpc)
self._setup_rpc()
self.ext_manager = self._create_agent_extension_manager(
self.connection)
# The initialization is complete; we can start receiving messages
self.connection.consume_in_threads()
# Initialize iteration counter
self.iter_num = 0
@ -111,7 +117,8 @@ class SriovNicSwitchAgent(object):
[topics.SECURITY_GROUP, topics.UPDATE]]
self.connection = agent_rpc.create_consumers(self.endpoints,
self.topic,
consumers)
consumers,
start_listening=False)
report_interval = cfg.CONF.AGENT.report_interval
if report_interval:
@ -129,8 +136,15 @@ class SriovNicSwitchAgent(object):
except Exception:
LOG.exception(_LE("Failed reporting state!"))
def _create_agent_extension_manager(self, connection):
ext_manager.register_opts(self.conf)
mgr = ext_manager.AgentExtensionsManager(self.conf)
mgr.initialize(connection, 'sriov')
return mgr
def setup_eswitch_mgr(self, device_mappings, exclude_devices={}):
self.eswitch_mgr = esm.ESwitchManager(device_mappings, exclude_devices)
self.eswitch_mgr = esm.ESwitchManager()
self.eswitch_mgr.discover_devices(device_mappings, exclude_devices)
def scan_devices(self, registered_devices, updated_devices):
curr_devices = self.eswitch_mgr.get_assigned_devices()
@ -224,6 +238,7 @@ class SriovNicSwitchAgent(object):
profile.get('pci_slot'),
device_details['admin_state_up'],
spoofcheck)
self.ext_manager.handle_port(self.context, device_details)
else:
LOG.info(_LI("Device with MAC %s not defined on plugin"),
device)
@ -234,6 +249,16 @@ class SriovNicSwitchAgent(object):
for device in devices:
LOG.info(_LI("Removing device with mac_address %s"), device)
try:
pci_slot = self.eswitch_mgr.get_pci_slot_by_mac(device)
if pci_slot:
profile = {'pci_slot': pci_slot}
port = {'device': device, 'profile': profile}
self.ext_manager.delete_port(self.context, port)
else:
LOG.warning(_LW("Failed to find pci slot for device "
"%(device)s; skipping extension port "
"cleanup"), device)
dev_details = self.plugin_rpc.update_device_down(self.context,
device,
self.agent_id,

View File

@ -24,6 +24,7 @@ from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers.mech_sriov.mech_driver \
import exceptions as exc
from neutron.services.qos import qos_consts
LOG = log.getLogger(__name__)
@ -61,6 +62,8 @@ class SriovNicSwitchMechanismDriver(api.MechanismDriver):
"""
supported_qos_rule_types = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT]
def __init__(self,
agent_type=constants.AGENT_TYPE_NIC_SWITCH,
vif_type=portbindings.VIF_TYPE_HW_VEB,

View File

@ -88,3 +88,5 @@ ARP_RESPONDER_ACTIONS = ('move:NXM_OF_ETH_SRC[]->NXM_OF_ETH_DST[],'
OVS_RESTARTED = 0
OVS_NORMAL = 1
OVS_DEAD = 2
EXTENSION_DRIVER_TYPE = 'ovs'

View File

@ -0,0 +1,76 @@
# 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
from neutron.i18n import _LW
from neutron.plugins.ml2.drivers.openvswitch.mech_driver import (
mech_openvswitch)
LOG = logging.getLogger(__name__)
class QosOVSAgentDriver(qos.QosAgentDriver):
_SUPPORTED_RULES = (
mech_openvswitch.OpenvswitchMechanismDriver.supported_qos_rule_types)
def __init__(self):
super(QosOVSAgentDriver, self).__init__()
self.br_int_name = cfg.CONF.OVS.integration_bridge
self.br_int = None
def initialize(self):
self.br_int = ovs_lib.OVSBridge(self.br_int_name)
def create(self, port, qos_policy):
self._handle_rules('create', port, qos_policy)
def update(self, port, qos_policy):
self._handle_rules('update', port, qos_policy)
def delete(self, port, qos_policy):
# TODO(QoS): consider optimizing flushing of all QoS rules from the
# port by inspecting qos_policy.rules contents
self._delete_bandwidth_limit(port)
def _handle_rules(self, action, port, qos_policy):
for rule in qos_policy.rules:
if rule.rule_type in self._SUPPORTED_RULES:
handler_name = ("".join(("_", action, "_", rule.rule_type)))
handler = getattr(self, handler_name)
handler(port, rule)
else:
LOG.warning(_LW('Unsupported QoS rule type for %(rule_id)s: '
'%(rule_type)s; skipping'),
{'rule_id': rule.id, 'rule_type': rule.rule_type})
def _create_bandwidth_limit(self, port, rule):
self._update_bandwidth_limit(port, rule)
def _update_bandwidth_limit(self, port, rule):
port_name = port['vif_port'].port_name
max_kbps = rule.max_kbps
max_burst_kbps = rule.max_burst_kbps
self.br_int.create_egress_bw_limit_for_port(port_name,
max_kbps,
max_burst_kbps)
def _delete_bandwidth_limit(self, port):
port_name = port['vif_port'].port_name
self.br_int.delete_egress_bw_limit_for_port(port_name)

View File

@ -29,6 +29,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.extensions import manager as ext_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
@ -224,6 +225,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_extension_manager(self.connection)
self.bridge_mappings = bridge_mappings
self.setup_physical_bridges(self.bridge_mappings)
self.local_vlan_map = {}
@ -364,6 +366,13 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
consumers,
start_listening=False)
def init_extension_manager(self, connection):
ext_manager.register_opts(self.conf)
self.ext_manager = (
ext_manager.AgentExtensionsManager(self.conf))
self.ext_manager.initialize(
connection, constants.EXTENSION_DRIVER_TYPE)
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:
@ -394,6 +403,9 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
# longer have access to the network
self.sg_agent.remove_devices_filter([port_id])
port = self.int_br.get_vif_port_by_id(port_id)
self.ext_manager.delete_port(self.context,
{"vif_port": port,
"port_id": port_id})
if port:
# don't log errors since there is a chance someone will be
# removing the port from the bridge at the same time
@ -1244,6 +1256,7 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
if 'port_id' in details:
LOG.info(_LI("Port %(device)s updated. Details: %(details)s"),
{'device': device, 'details': details})
details['vif_port'] = port
need_binding = self.treat_vif_port(port, details['port_id'],
details['network_id'],
details['network_type'],
@ -1254,8 +1267,8 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
details['device_owner'],
ovs_restarted)
if need_binding:
details['vif_port'] = port
need_binding_devices.append(details)
self.ext_manager.handle_port(self.context, details)
else:
LOG.warn(_LW("Device %s not defined on plugin"), device)
if (port and port.ofport != -1):

View File

@ -20,6 +20,7 @@ from neutron.common import constants
from neutron.extensions import portbindings
from neutron.plugins.common import constants as p_constants
from neutron.plugins.ml2.drivers import mech_agent
from neutron.services.qos import qos_consts
LOG = log.getLogger(__name__)
@ -34,6 +35,8 @@ class OpenvswitchMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
network.
"""
supported_qos_rule_types = [qos_consts.RULE_TYPE_BANDWIDTH_LIMIT]
def __init__(self):
sg_enabled = securitygroups_rpc.is_firewall_enabled()
vif_details = {portbindings.CAP_PORT_FILTER: sg_enabled,

View 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.core_extensions import base as base_core
from neutron.core_extensions import qos as qos_core
from neutron.plugins.ml2 import driver_api as api
LOG = logging.getLogger(__name__)
class QosExtensionDriver(api.ExtensionDriver):
def initialize(self):
self.core_ext_handler = qos_core.QosCoreResourceExtension()
LOG.debug("QosExtensionDriver initialization complete")
def process_create_network(self, context, data, result):
self.core_ext_handler.process_fields(
context, base_core.NETWORK, data, result)
process_update_network = process_create_network
def process_create_port(self, context, data, result):
self.core_ext_handler.process_fields(
context, base_core.PORT, data, result)
process_update_port = process_create_port
def extend_network_dict(self, session, db_data, result):
result.update(
self.core_ext_handler.extract_fields(
base_core.NETWORK, db_data))
def extend_port_dict(self, session, db_data, result):
result.update(
self.core_ext_handler.extract_fields(base_core.PORT, db_data))

View File

@ -26,11 +26,12 @@ from neutron.extensions import multiprovidernet as mpnet
from neutron.extensions import portbindings
from neutron.extensions import providernet as provider
from neutron.extensions import vlantransparent
from neutron.i18n import _LE, _LI
from neutron.i18n import _LE, _LI, _LW
from neutron.plugins.ml2.common import exceptions as ml2_exc
from neutron.plugins.ml2 import db
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2 import models
from neutron.services.qos import qos_consts
LOG = log.getLogger(__name__)
@ -313,6 +314,40 @@ class MechanismManager(stevedore.named.NamedExtensionManager):
LOG.info(_LI("Registered mechanism drivers: %s"),
[driver.name for driver in self.ordered_mech_drivers])
@property
def supported_qos_rule_types(self):
if not self.ordered_mech_drivers:
return []
rule_types = set(qos_consts.VALID_RULE_TYPES)
# Recalculate on every call to allow drivers determine supported rule
# types dynamically
for driver in self.ordered_mech_drivers:
if hasattr(driver.obj, 'supported_qos_rule_types'):
new_rule_types = \
rule_types & set(driver.obj.supported_qos_rule_types)
dropped_rule_types = new_rule_types - rule_types
if dropped_rule_types:
LOG.info(
_LI("%(rule_types)s rule types disabled for ml2 "
"because %(driver)s does not support them"),
{'rule_types': ', '.join(dropped_rule_types),
'driver': driver.name})
rule_types = new_rule_types
else:
# at least one of drivers does not support QoS, meaning there
# are no rule types supported by all of them
LOG.warn(
_LW("%s does not support QoS; no rule types available"),
driver.name)
return []
rule_types = list(rule_types)
LOG.debug("Supported QoS rule types "
"(common subset for all mech drivers): %s", rule_types)
return rule_types
def initialize(self):
for driver in self.ordered_mech_drivers:
LOG.info(_LI("Initializing mechanism driver '%s'"), driver.name)
@ -754,9 +789,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):

View File

@ -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
@ -76,6 +77,7 @@ from neutron.plugins.ml2 import managers
from neutron.plugins.ml2 import models
from neutron.plugins.ml2 import rpc
from neutron.quota import resource_registry
from neutron.services.qos import qos_consts
LOG = log.getLogger(__name__)
@ -161,7 +163,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.ResourcesPullRpcCallback()
]
def _setup_dhcp(self):
@ -171,6 +174,10 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
)
self.start_periodic_dhcp_agent_status_check()
@property
def supported_qos_rule_types(self):
return self.mechanism_manager.supported_qos_rule_types
@log_helpers.log_method_call
def start_rpc_listeners(self):
"""Start the RPC loop to let the plugin communicate with agents."""
@ -1119,6 +1126,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_consts.QOS_POLICY_ID in attrs and
original_port[qos_consts.QOS_POLICY_ID] !=
updated_port[qos_consts.QOS_POLICY_ID]):
need_port_update_notify = True
if addr_pair.ADDRESS_PAIRS in attrs:
need_port_update_notify |= (

View File

@ -32,6 +32,7 @@ from neutron.i18n import _LE, _LW
from neutron import manager
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_tunnel
from neutron.services.qos import qos_consts
# REVISIT(kmestery): Allow the type and mechanism drivers to supply the
# mixins and eventually remove the direct dependencies on type_tunnel.
@ -108,6 +109,9 @@ class RpcCallbacks(type_tunnel.TunnelRpcCallbackMixin):
host,
port_context.network.current)
qos_policy_id = (port.get(qos_consts.QOS_POLICY_ID) or
port_context.network._network.get(
qos_consts.QOS_POLICY_ID))
entry = {'device': device,
'network_id': port['network_id'],
'port_id': port['id'],
@ -120,6 +124,7 @@ class RpcCallbacks(type_tunnel.TunnelRpcCallbackMixin):
'device_owner': port['device_owner'],
'allowed_address_pairs': port['allowed_address_pairs'],
'port_security_enabled': port.get(psec.PORTSECURITY, True),
'qos_policy_id': qos_policy_id,
'profile': port[portbindings.PROFILE]}
LOG.debug("Returning: %s", entry)
return entry

View File

View File

@ -0,0 +1,74 @@
# 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.i18n import _LI
from neutron import manager
QOS_DRIVER_NAMESPACE = 'neutron.qos.notification_drivers'
QOS_PLUGIN_OPTS = [
cfg.ListOpt('notification_drivers',
default='message_queue',
help=_('Drivers list to use to send the update notification')),
]
cfg.CONF.register_opts(QOS_PLUGIN_OPTS, "qos")
LOG = logging.getLogger(__name__)
class QosServiceNotificationDriverManager(object):
def __init__(self):
self.notification_drivers = []
self._load_drivers(cfg.CONF.qos.notification_drivers)
def update_policy(self, context, qos_policy):
for driver in self.notification_drivers:
driver.update_policy(context, qos_policy)
def delete_policy(self, context, qos_policy):
for driver in self.notification_drivers:
driver.delete_policy(context, qos_policy)
def create_policy(self, context, qos_policy):
for driver in self.notification_drivers:
driver.create_policy(context, qos_policy)
def _load_drivers(self, notification_drivers):
"""Load all the instances of the configured QoS notification drivers
:param notification_drivers: comma separated string
"""
if not notification_drivers:
raise SystemExit(_('A QoS driver must be specified'))
LOG.debug("Loading QoS notification drivers: %s", notification_drivers)
for notification_driver in notification_drivers:
driver_ins = self._load_driver_instance(notification_driver)
self.notification_drivers.append(driver_ins)
def _load_driver_instance(self, notification_driver):
"""Returns an instance of the configured QoS notification driver
:returns: An instance of Driver for the QoS notification
"""
mgr = manager.NeutronManager
driver = mgr.load_class_for_provider(QOS_DRIVER_NAMESPACE,
notification_driver)
driver_instance = driver()
LOG.info(
_LI("Loading %(name)s (%(description)s) notification driver "
"for QoS plugin"),
{"name": notification_driver,
"description": driver_instance.get_description()})
return driver_instance

View File

@ -0,0 +1,59 @@
# 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.api.rpc.callbacks import events
from neutron.api.rpc.callbacks.producer import registry
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron.i18n import _LW
from neutron.objects.qos import policy as policy_object
from neutron.services.qos.notification_drivers import qos_base
LOG = logging.getLogger(__name__)
def _get_qos_policy_cb(resource, policy_id, **kwargs):
context = kwargs.get('context')
if context is None:
LOG.warning(_LW(
'Received %(resource)s %(policy_id)s without context'),
{'resource': resource, 'policy_id': policy_id}
)
return
policy = policy_object.QosPolicy.get_by_id(context, policy_id)
return policy
class RpcQosServiceNotificationDriver(
qos_base.QosServiceNotificationDriverBase):
"""RPC message queue service notification driver for QoS."""
def __init__(self):
self.notification_api = resources_rpc.ResourcesPushRpcApi()
registry.provide(_get_qos_policy_cb, resources.QOS_POLICY)
def get_description(self):
return "Message queue updates"
def create_policy(self, context, policy):
#No need to update agents on create
pass
def update_policy(self, context, policy):
self.notification_api.push(context, policy, events.UPDATED)
def delete_policy(self, context, policy):
self.notification_api.push(context, policy, events.DELETED)

View File

@ -0,0 +1,42 @@
# 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 QosServiceNotificationDriverBase(object):
"""QoS service notification driver base class."""
@abc.abstractmethod
def get_description(self):
"""Get the notification driver description.
"""
@abc.abstractmethod
def create_policy(self, context, policy):
"""Create the QoS policy."""
@abc.abstractmethod
def update_policy(self, context, policy):
"""Update the QoS policy.
Apply changes to the QoS policy.
"""
@abc.abstractmethod
def delete_policy(self, context, policy):
"""Delete the QoS policy.
Remove all rules for this policy and free up all the resources.
"""

View File

@ -0,0 +1,19 @@
# 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.
RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth_limit'
VALID_RULE_TYPES = [RULE_TYPE_BANDWIDTH_LIMIT]
QOS_POLICY_ID = 'qos_policy_id'

View File

@ -0,0 +1,163 @@
# 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.common import exceptions as n_exc
from neutron.db import api as db_api
from neutron.db import db_base_plugin_common
from neutron.extensions import qos
from neutron.objects.qos import policy as policy_object
from neutron.objects.qos import rule as rule_object
from neutron.objects.qos import rule_type as rule_type_object
from neutron.services.qos.notification_drivers import manager as driver_mgr
LOG = logging.getLogger(__name__)
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.notification_driver_manager = (
driver_mgr.QosServiceNotificationDriverManager())
@db_base_plugin_common.convert_result_to_dict
def create_policy(self, context, policy):
policy = policy_object.QosPolicy(context, **policy['policy'])
policy.create()
self.notification_driver_manager.create_policy(context, policy)
return policy
@db_base_plugin_common.convert_result_to_dict
def update_policy(self, context, policy_id, policy):
policy = policy_object.QosPolicy(context, **policy['policy'])
policy.id = policy_id
policy.update()
self.notification_driver_manager.update_policy(context, policy)
return policy
def delete_policy(self, context, policy_id):
policy = policy_object.QosPolicy(context)
policy.id = policy_id
self.notification_driver_manager.delete_policy(context, policy)
policy.delete()
def _get_policy_obj(self, context, policy_id):
obj = policy_object.QosPolicy.get_by_id(context, policy_id)
if obj is None:
raise n_exc.QosPolicyNotFound(policy_id=policy_id)
return obj
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy(self, context, policy_id, fields=None):
return self._get_policy_obj(context, policy_id)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policies(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
return policy_object.QosPolicy.get_objects(context, **filters)
#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.
@db_base_plugin_common.convert_result_to_dict
def create_policy_bandwidth_limit_rule(self, context, policy_id,
bandwidth_limit_rule):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule(
context, qos_policy_id=policy_id,
**bandwidth_limit_rule['bandwidth_limit_rule'])
rule.create()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
@db_base_plugin_common.convert_result_to_dict
def update_policy_bandwidth_limit_rule(self, context, rule_id, policy_id,
bandwidth_limit_rule):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule(
context, **bandwidth_limit_rule['bandwidth_limit_rule'])
rule.id = rule_id
rule.update()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
return rule
def delete_policy_bandwidth_limit_rule(self, context, rule_id, policy_id):
# make sure we will have a policy object to push resource update
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
policy = self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule(context)
rule.id = rule_id
rule.delete()
policy.reload_rules()
self.notification_driver_manager.update_policy(context, policy)
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_policy_bandwidth_limit_rule(self, context, rule_id,
policy_id, fields=None):
# make sure we have access to the policy when fetching the rule
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
rule = rule_object.QosBandwidthLimitRule.get_by_id(
context, rule_id)
if not rule:
raise n_exc.QosRuleNotFound(policy_id=policy_id, rule_id=rule_id)
return rule
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_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):
# make sure we have access to the policy when fetching rules
with db_api.autonested_transaction(context.session):
# first, validate that we have access to the policy
self._get_policy_obj(context, policy_id)
return rule_object.QosBandwidthLimitRule.get_objects(context,
**filters)
# TODO(QoS): enforce rule types when accessing rule objects
@db_base_plugin_common.filter_fields
@db_base_plugin_common.convert_result_to_dict
def get_rule_types(self, context, filters=None, fields=None,
sorts=None, limit=None,
marker=None, page_reverse=False):
return rule_type_object.QosRuleType.get_objects(**filters)

View File

@ -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 = []
@ -115,6 +117,14 @@ class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
for vpnservice in cls.vpnservices:
cls._try_delete_resource(cls.client.delete_vpnservice,
vpnservice['id'])
# Clean up QoS rules
for qos_rule in cls.qos_rules:
cls._try_delete_resource(cls.admin_client.delete_qos_rule,
qos_rule['id'])
# Clean up QoS policies
for qos_policy in cls.qos_policies:
cls._try_delete_resource(cls.admin_client.delete_qos_policy,
qos_policy['id'])
# Clean up floating IPs
for floating_ip in cls.floating_ips:
cls._try_delete_resource(cls.client.delete_floatingip,
@ -221,9 +231,9 @@ class BaseNetworkTest(neutron.tests.tempest.test.BaseTestCase):
return network
@classmethod
def create_shared_network(cls, network_name=None):
def create_shared_network(cls, network_name=None, **post_body):
network_name = network_name or data_utils.rand_name('sharednetwork-')
post_body = {'name': network_name, 'shared': True}
post_body.update({'name': network_name, 'shared': True})
body = cls.admin_client.create_network(**post_body)
network = body['network']
cls.shared_networks.append(network)
@ -431,6 +441,25 @@ 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, tenant_id=None):
"""Wrapper utility that returns a test QoS policy."""
body = cls.admin_client.create_qos_policy(
name, description, shared, tenant_id)
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.admin_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'])

View File

@ -0,0 +1,402 @@
# 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 tempest_lib import exceptions
from neutron.services.qos import qos_consts
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 desc1',
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 desc1', retrieved_policy['description'])
self.assertFalse(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('f8d20e92-f06d-4805-b54f-230f77715815')
def test_list_policy_filter_by_name(self):
self.create_qos_policy(name='test', description='test policy',
shared=False)
self.create_qos_policy(name='test2', description='test policy',
shared=False)
policies = (self.admin_client.
list_qos_policies(name='test')['policies'])
self.assertEqual(1, len(policies))
retrieved_policy = policies[0]
self.assertEqual('test', retrieved_policy['name'])
@test.attr(type='smoke')
@test.idempotent_id('8e88a54b-f0b2-4b7d-b061-a15d93c2c7d6')
def test_policy_update(self):
policy = self.create_qos_policy(name='test-policy',
description='',
shared=False)
self.admin_client.update_qos_policy(policy['id'],
description='test policy desc2',
shared=True)
retrieved_policy = self.admin_client.show_qos_policy(policy['id'])
retrieved_policy = retrieved_policy['policy']
self.assertEqual('test policy desc2', retrieved_policy['description'])
self.assertTrue(retrieved_policy['shared'])
self.assertEqual([], retrieved_policy['rules'])
@test.attr(type='smoke')
@test.idempotent_id('1cb42653-54bd-4a9a-b888-c55e18199201')
def test_delete_policy(self):
policy = self.admin_client.create_qos_policy(
'test-policy', 'desc', True)['policy']
retrieved_policy = self.admin_client.show_qos_policy(policy['id'])
retrieved_policy = retrieved_policy['policy']
self.assertEqual('test-policy', retrieved_policy['name'])
self.admin_client.delete_qos_policy(policy['id'])
self.assertRaises(exceptions.NotFound,
self.admin_client.show_qos_policy, policy['id'])
@test.attr(type='smoke')
@test.idempotent_id('cf776f77-8d3d-49f2-8572-12d6a1557224')
def test_list_rule_types(self):
# List supported rule types
# TODO(QoS): since in gate we run both ovs and linuxbridge ml2 drivers,
# and since Linux Bridge ml2 driver does not have QoS support yet, ml2
# plugin reports no rule types are supported. Once linuxbridge will
# receive support for QoS, the list of expected rule types will change.
#
# In theory, we could make the test conditional on which ml2 drivers
# are enabled in gate (or more specifically, on which supported qos
# rules are claimed by core plugin), but that option doesn't seem to be
# available thru tempest_lib framework
expected_rule_types = []
expected_rule_details = ['type']
rule_types = self.admin_client.list_qos_rule_types()
actual_list_rule_types = rule_types['rule_types']
actual_rule_types = [rule['type'] for rule in actual_list_rule_types]
# Verify that only required fields present in rule details
for rule in actual_list_rule_types:
self.assertEqual(tuple(rule.keys()), tuple(expected_rule_details))
# Verify if expected rules are present in the actual rules list
for rule in expected_rule_types:
self.assertIn(rule, actual_rule_types)
def _disassociate_network(self, client, network_id):
client.update_network(network_id, qos_policy_id=None)
updated_network = self.admin_client.show_network(network_id)
self.assertIsNone(updated_network['network']['qos_policy_id'])
@test.attr(type='smoke')
@test.idempotent_id('65b9ef75-1911-406a-bbdb-ca1d68d528b0')
def test_policy_association_with_admin_network(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
network = self.create_shared_network('test network',
qos_policy_id=policy['id'])
retrieved_network = self.admin_client.show_network(network['id'])
self.assertEqual(
policy['id'], retrieved_network['network']['qos_policy_id'])
self._disassociate_network(self.admin_client, network['id'])
@test.attr(type='smoke')
@test.idempotent_id('1738de5d-0476-4163-9022-5e1b548c208e')
def test_policy_association_with_tenant_network(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=True)
network = self.create_network('test network',
qos_policy_id=policy['id'])
retrieved_network = self.admin_client.show_network(network['id'])
self.assertEqual(
policy['id'], retrieved_network['network']['qos_policy_id'])
self._disassociate_network(self.client, network['id'])
@test.attr(type='smoke')
@test.idempotent_id('9efe63d0-836f-4cc2-b00c-468e63aa614e')
def test_policy_association_with_network_nonexistent_policy(self):
self.assertRaises(
exceptions.NotFound,
self.create_network,
'test network',
qos_policy_id='9efe63d0-836f-4cc2-b00c-468e63aa614e')
@test.attr(type='smoke')
@test.idempotent_id('1aa55a79-324f-47d9-a076-894a8fc2448b')
def test_policy_association_with_network_non_shared_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
self.assertRaises(
exceptions.NotFound,
self.create_network,
'test network', qos_policy_id=policy['id'])
@test.attr(type='smoke')
@test.idempotent_id('09a9392c-1359-4cbb-989f-fb768e5834a8')
def test_policy_update_association_with_admin_network(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
network = self.create_shared_network('test network')
retrieved_network = self.admin_client.show_network(network['id'])
self.assertIsNone(retrieved_network['network']['qos_policy_id'])
self.admin_client.update_network(network['id'],
qos_policy_id=policy['id'])
retrieved_network = self.admin_client.show_network(network['id'])
self.assertEqual(
policy['id'], retrieved_network['network']['qos_policy_id'])
self._disassociate_network(self.admin_client, network['id'])
def _disassociate_port(self, port_id):
self.client.update_port(port_id, qos_policy_id=None)
updated_port = self.admin_client.show_port(port_id)
self.assertIsNone(updated_port['port']['qos_policy_id'])
@test.attr(type='smoke')
@test.idempotent_id('98fcd95e-84cf-4746-860e-44692e674f2e')
def test_policy_association_with_port_shared_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=True)
network = self.create_shared_network('test network')
port = self.create_port(network, qos_policy_id=policy['id'])
retrieved_port = self.admin_client.show_port(port['id'])
self.assertEqual(
policy['id'], retrieved_port['port']['qos_policy_id'])
self._disassociate_port(port['id'])
@test.attr(type='smoke')
@test.idempotent_id('49e02f5a-e1dd-41d5-9855-cfa37f2d195e')
def test_policy_association_with_port_nonexistent_policy(self):
network = self.create_shared_network('test network')
self.assertRaises(
exceptions.NotFound,
self.create_port,
network,
qos_policy_id='49e02f5a-e1dd-41d5-9855-cfa37f2d195e')
@test.attr(type='smoke')
@test.idempotent_id('f53d961c-9fe5-4422-8b66-7add972c6031')
def test_policy_association_with_port_non_shared_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
network = self.create_shared_network('test network')
self.assertRaises(
exceptions.NotFound,
self.create_port,
network, qos_policy_id=policy['id'])
@test.attr(type='smoke')
@test.idempotent_id('f8163237-fba9-4db5-9526-bad6d2343c76')
def test_policy_update_association_with_port_shared_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=True)
network = self.create_shared_network('test network')
port = self.create_port(network)
retrieved_port = self.admin_client.show_port(port['id'])
self.assertIsNone(retrieved_port['port']['qos_policy_id'])
self.client.update_port(port['id'], qos_policy_id=policy['id'])
retrieved_port = self.admin_client.show_port(port['id'])
self.assertEqual(
policy['id'], retrieved_port['port']['qos_policy_id'])
self._disassociate_port(port['id'])
@test.attr(type='smoke')
@test.idempotent_id('18163237-8ba9-4db5-9525-bad6d2343c75')
def test_delete_not_allowed_if_policy_in_use_by_network(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=True)
network = self.create_shared_network(
'test network', qos_policy_id=policy['id'])
self.assertRaises(
exceptions.Conflict,
self.admin_client.delete_qos_policy, policy['id'])
self._disassociate_network(self.admin_client, network['id'])
self.admin_client.delete_qos_policy(policy['id'])
@test.attr(type='smoke')
@test.idempotent_id('24153230-84a9-4dd5-9525-bad6d2343c75')
def test_delete_not_allowed_if_policy_in_use_by_port(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=True)
network = self.create_shared_network('test network')
port = self.create_port(network, qos_policy_id=policy['id'])
self.assertRaises(
exceptions.Conflict,
self.admin_client.delete_qos_policy, policy['id'])
self._disassociate_port(port['id'])
self.admin_client.delete_qos_policy(policy['id'])
class QosBandwidthLimitRuleTestJSON(base.BaseAdminNetworkTest):
@classmethod
def resource_setup(cls):
super(QosBandwidthLimitRuleTestJSON, 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('8a59b00b-3e9c-4787-92f8-93a5cdf5e378')
def test_rule_create(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_rule = self.admin_client.show_bandwidth_limit_rule(
policy['id'], rule['id'])
retrieved_rule = retrieved_rule['bandwidth_limit_rule']
self.assertEqual(rule['id'], retrieved_rule['id'])
self.assertEqual(200, retrieved_rule['max_kbps'])
self.assertEqual(1337, retrieved_rule['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)
# Test 'show policy'
retrieved_policy = self.admin_client.show_qos_policy(policy['id'])
policy_rules = retrieved_policy['policy']['rules']
self.assertEqual(1, len(policy_rules))
self.assertEqual(rule['id'], policy_rules[0]['id'])
self.assertEqual(qos_consts.RULE_TYPE_BANDWIDTH_LIMIT,
policy_rules[0]['type'])
@test.attr(type='smoke')
@test.idempotent_id('8a59b00b-ab01-4787-92f8-93a5cdf5e378')
def test_rule_create_fail_for_the_same_type(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
self.create_qos_bandwidth_limit_rule(policy_id=policy['id'],
max_kbps=200,
max_burst_kbps=1337)
self.assertRaises(exceptions.Conflict,
self.create_qos_bandwidth_limit_rule,
policy_id=policy['id'],
max_kbps=201, max_burst_kbps=1338)
@test.attr(type='smoke')
@test.idempotent_id('149a6988-2568-47d2-931e-2dbc858943b3')
def test_rule_update(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=1,
max_burst_kbps=1)
self.admin_client.update_bandwidth_limit_rule(policy['id'],
rule['id'],
max_kbps=200,
max_burst_kbps=1337)
retrieved_policy = self.admin_client.show_bandwidth_limit_rule(
policy['id'], rule['id'])
retrieved_policy = retrieved_policy['bandwidth_limit_rule']
self.assertEqual(200, retrieved_policy['max_kbps'])
self.assertEqual(1337, retrieved_policy['max_burst_kbps'])
@test.attr(type='smoke')
@test.idempotent_id('67ee6efd-7b33-4a68-927d-275b4f8ba958')
def test_rule_delete(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False)
rule = self.admin_client.create_bandwidth_limit_rule(
policy['id'], 200, 1337)['bandwidth_limit_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.admin_client.delete_bandwidth_limit_rule(policy['id'], rule['id'])
self.assertRaises(exceptions.NotFound,
self.admin_client.show_bandwidth_limit_rule,
policy['id'], rule['id'])
@test.attr(type='smoke')
@test.idempotent_id('f211222c-5808-46cb-a961-983bbab6b852')
def test_rule_create_rule_nonexistent_policy(self):
self.assertRaises(
exceptions.NotFound,
self.create_qos_bandwidth_limit_rule,
'policy', 200, 1337)
@test.attr(type='smoke')
@test.idempotent_id('3ba4abf9-7976-4eaf-a5d0-a934a6e09b2d')
def test_rule_association_nonshared_policy(self):
policy = self.create_qos_policy(name='test-policy',
description='test policy',
shared=False,
tenant_id='tenant-id')
self.assertRaises(
exceptions.NotFound,
self.client.create_bandwidth_limit_rule,
policy['id'], 200, 1337)

View File

@ -35,6 +35,7 @@ import six
import testtools
from neutron.agent.linux import external_process
from neutron.api.rpc.callbacks.consumer import registry as rpc_consumer_reg
from neutron.callbacks import manager as registry_manager
from neutron.callbacks import registry
from neutron.common import config
@ -290,6 +291,7 @@ class BaseTestCase(DietTestCase):
policy.init()
self.addCleanup(policy.reset)
self.addCleanup(rpc_consumer_reg.clear)
def get_new_temp_dir(self):
"""Create a new temporary directory.

View File

@ -0,0 +1,27 @@
# 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.agent.linux import utils as agent_utils
def wait_until_bandwidth_limit_rule_applied(bridge, port_vif, rule):
def _bandwidth_limit_rule_applied():
bw_rule = bridge.get_egress_bw_limit_for_port(port_vif)
expected = None, None
if rule:
expected = rule.max_kbps, rule.max_burst_kbps
return bw_rule == expected
agent_utils.wait_until_true(_bandwidth_limit_rule_applied)

View File

@ -41,5 +41,7 @@ EOF
enable_plugin neutron-vpnaas git://git.openstack.org/openstack/neutron-vpnaas
"
export DEVSTACK_LOCAL_CONFIG+="DISABLE_NETWORK_API_EXTENSIONS=qos
"
$BASE/new/devstack-gate/devstack-vm-gate.sh
fi

View File

@ -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",

View File

@ -0,0 +1,286 @@
# Copyright (c) 2015 Red Hat, Inc.
# Copyright (c) 2015 SUSE Linux Products GmbH
# 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 random
import eventlet
import mock
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import uuidutils
from neutron.agent.common import config as agent_config
from neutron.agent.common import ovs_lib
from neutron.agent.l2.extensions import manager as ext_manager
from neutron.agent.linux import interface
from neutron.agent.linux import polling
from neutron.agent.linux import utils as agent_utils
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2.drivers.openvswitch.agent.common import config \
as ovs_config
from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_int
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_phys
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_tun
from neutron.plugins.ml2.drivers.openvswitch.agent import ovs_neutron_agent \
as ovs_agent
from neutron.tests.functional.agent.linux import base
LOG = logging.getLogger(__name__)
class OVSAgentTestFramework(base.BaseOVSLinuxTestCase):
def setUp(self):
super(OVSAgentTestFramework, self).setUp()
agent_rpc = ('neutron.plugins.ml2.drivers.openvswitch.agent.'
'ovs_neutron_agent.OVSPluginApi')
mock.patch(agent_rpc).start()
mock.patch('neutron.agent.rpc.PluginReportStateAPI').start()
self.br_int = base.get_rand_name(n_const.DEVICE_NAME_MAX_LEN,
prefix='br-int')
self.br_tun = base.get_rand_name(n_const.DEVICE_NAME_MAX_LEN,
prefix='br-tun')
patch_name_len = n_const.DEVICE_NAME_MAX_LEN - len("-patch-tun")
self.patch_tun = "%s-patch-tun" % self.br_int[patch_name_len:]
self.patch_int = "%s-patch-int" % self.br_tun[patch_name_len:]
self.ovs = ovs_lib.BaseOVS()
self.config = self._configure_agent()
self.driver = interface.OVSInterfaceDriver(self.config)
def _get_config_opts(self):
config = cfg.ConfigOpts()
config.register_opts(common_config.core_opts)
config.register_opts(interface.OPTS)
config.register_opts(ovs_config.ovs_opts, "OVS")
config.register_opts(ovs_config.agent_opts, "AGENT")
agent_config.register_interface_driver_opts_helper(config)
agent_config.register_agent_state_opts_helper(config)
ext_manager.register_opts(config)
return config
def _configure_agent(self):
config = self._get_config_opts()
config.set_override(
'interface_driver',
'neutron.agent.linux.interface.OVSInterfaceDriver')
config.set_override('integration_bridge', self.br_int, "OVS")
config.set_override('ovs_integration_bridge', self.br_int)
config.set_override('tunnel_bridge', self.br_tun, "OVS")
config.set_override('int_peer_patch_port', self.patch_tun, "OVS")
config.set_override('tun_peer_patch_port', self.patch_int, "OVS")
config.set_override('host', 'ovs-agent')
return config
def _bridge_classes(self):
return {
'br_int': br_int.OVSIntegrationBridge,
'br_phys': br_phys.OVSPhysicalBridge,
'br_tun': br_tun.OVSTunnelBridge
}
def create_agent(self, create_tunnels=True):
if create_tunnels:
tunnel_types = [p_const.TYPE_VXLAN]
else:
tunnel_types = None
local_ip = '192.168.10.1'
bridge_mappings = {'physnet': self.br_int}
agent = ovs_agent.OVSNeutronAgent(self._bridge_classes(),
self.br_int, self.br_tun,
local_ip, bridge_mappings,
polling_interval=1,
tunnel_types=tunnel_types,
prevent_arp_spoofing=False,
conf=self.config)
self.addCleanup(self.ovs.delete_bridge, self.br_int)
if tunnel_types:
self.addCleanup(self.ovs.delete_bridge, self.br_tun)
agent.sg_agent = mock.Mock()
return agent
def start_agent(self, agent):
self.setup_agent_rpc_mocks(agent)
polling_manager = polling.InterfacePollingMinimizer()
self.addCleanup(polling_manager.stop)
polling_manager.start()
agent_utils.wait_until_true(
polling_manager._monitor.is_active)
agent.check_ovs_status = mock.Mock(
return_value=constants.OVS_NORMAL)
t = eventlet.spawn(agent.rpc_loop, polling_manager)
def stop_agent(agent, rpc_loop_thread):
agent.run_daemon_loop = False
rpc_loop_thread.wait()
self.addCleanup(stop_agent, agent, t)
def _bind_ports(self, ports, network, agent):
devices = []
for port in ports:
dev = OVSAgentTestFramework._get_device_details(port, network)
vif_name = port.get('vif_name')
vif_id = uuidutils.generate_uuid(),
vif_port = ovs_lib.VifPort(
vif_name, "%s" % vif_id, 'id-%s' % vif_id,
port.get('mac_address'), agent.int_br)
dev['vif_port'] = vif_port
devices.append(dev)
agent._bind_devices(devices)
def _create_test_port_dict(self):
return {'id': uuidutils.generate_uuid(),
'mac_address': utils.get_random_mac(
'fa:16:3e:00:00:00'.split(':')),
'fixed_ips': [{
'ip_address': '10.%d.%d.%d' % (
random.randint(3, 254),
random.randint(3, 254),
random.randint(3, 254))}],
'vif_name': base.get_rand_name(
self.driver.DEV_NAME_LEN, self.driver.DEV_NAME_PREFIX)}
def _create_test_network_dict(self):
return {'id': uuidutils.generate_uuid(),
'tenant_id': uuidutils.generate_uuid()}
def _plug_ports(self, network, ports, agent, ip_len=24):
for port in ports:
self.driver.plug(
network.get('id'), port.get('id'), port.get('vif_name'),
port.get('mac_address'),
agent.int_br.br_name, namespace=None)
ip_cidrs = ["%s/%s" % (port.get('fixed_ips')[0][
'ip_address'], ip_len)]
self.driver.init_l3(port.get('vif_name'), ip_cidrs, namespace=None)
def _get_device_details(self, port, network):
dev = {'device': port['id'],
'port_id': port['id'],
'network_id': network['id'],
'network_type': 'vlan',
'physical_network': 'physnet',
'segmentation_id': 1,
'fixed_ips': port['fixed_ips'],
'device_owner': 'compute',
'admin_state_up': True}
return dev
def assert_bridge(self, br, exists=True):
self.assertEqual(exists, self.ovs.bridge_exists(br))
def assert_patch_ports(self, agent):
def get_peer(port):
return agent.int_br.db_get_val(
'Interface', port, 'options', check_error=True)
agent_utils.wait_until_true(
lambda: get_peer(self.patch_int) == {'peer': self.patch_tun})
agent_utils.wait_until_true(
lambda: get_peer(self.patch_tun) == {'peer': self.patch_int})
def assert_bridge_ports(self):
for port in [self.patch_tun, self.patch_int]:
self.assertTrue(self.ovs.port_exists(port))
def assert_vlan_tags(self, ports, agent):
for port in ports:
res = agent.int_br.db_get_val('Port', port.get('vif_name'), 'tag')
self.assertTrue(res)
def _expected_plugin_rpc_call(self, call, expected_devices, is_up=True):
"""Helper to check expected rpc call are received
:param call: The call to check
:param expected_devices The device for which call is expected
:param is_up True if expected_devices are devices that are set up,
False if expected_devices are devices that are set down
"""
if is_up:
rpc_devices = [
dev for args in call.call_args_list for dev in args[0][1]]
else:
rpc_devices = [
dev for args in call.call_args_list for dev in args[0][2]]
return not (set(expected_devices) - set(rpc_devices))
def create_test_ports(self, amount=3, **kwargs):
ports = []
for x in range(amount):
ports.append(self._create_test_port_dict(**kwargs))
return ports
def _mock_update_device(self, context, devices_up, devices_down, agent_id,
host=None):
dev_up = []
dev_down = []
for port in self.ports:
if devices_up and port['id'] in devices_up:
dev_up.append(port['id'])
if devices_down and port['id'] in devices_down:
dev_down.append({'device': port['id'], 'exists': True})
return {'devices_up': dev_up,
'failed_devices_up': [],
'devices_down': dev_down,
'failed_devices_down': []}
def setup_agent_rpc_mocks(self, agent):
def mock_device_details(context, devices, agent_id, host=None):
details = []
for port in self.ports:
if port['id'] in devices:
dev = self._get_device_details(
port, self.network)
details.append(dev)
return {'devices': details, 'failed_devices': []}
(agent.plugin_rpc.get_devices_details_list_and_failed_devices.
side_effect) = mock_device_details
agent.plugin_rpc.update_device_list.side_effect = (
self._mock_update_device)
def _prepare_resync_trigger(self, agent):
def mock_device_raise_exception(context, devices_up, devices_down,
agent_id, host=None):
agent.plugin_rpc.update_device_list.side_effect = (
self._mock_update_device)
raise Exception('Exception to trigger resync')
self.agent.plugin_rpc.update_device_list.side_effect = (
mock_device_raise_exception)
def wait_until_ports_state(self, ports, up):
port_ids = [p['id'] for p in ports]
agent_utils.wait_until_true(
lambda: self._expected_plugin_rpc_call(
self.agent.plugin_rpc.update_device_list, port_ids, up))
def setup_agent_and_ports(self, port_dicts, trigger_resync=False):
self.agent = self.create_agent()
self.start_agent(self.agent)
self.network = self._create_test_network_dict()
self.ports = port_dicts
if trigger_resync:
self._prepare_resync_trigger(self.agent)
self._plug_ports(self.network, self.ports, self.agent)

View 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.
import copy
import mock
from oslo_utils import uuidutils
from neutron.api.rpc.callbacks.consumer import registry as consumer_reg
from neutron.api.rpc.callbacks import events
from neutron.api.rpc.callbacks import resources
from neutron.objects.qos import policy
from neutron.objects.qos import rule
from neutron.tests.common.agents import l2_extensions
from neutron.tests.functional.agent.l2 import base
TEST_POLICY_ID1 = "a2d72369-4246-4f19-bd3c-af51ec8d70cd"
TEST_POLICY_ID2 = "46ebaec0-0570-43ac-82f6-60d2b03168c5"
TEST_BW_LIMIT_RULE_1 = rule.QosBandwidthLimitRule(
context=None,
id="5f126d84-551a-4dcf-bb01-0e9c0df0c793",
max_kbps=1000,
max_burst_kbps=10)
TEST_BW_LIMIT_RULE_2 = rule.QosBandwidthLimitRule(
context=None,
id="fa9128d9-44af-49b2-99bb-96548378ad42",
max_kbps=900,
max_burst_kbps=9)
class OVSAgentQoSExtensionTestFramework(base.OVSAgentTestFramework):
def setUp(self):
super(OVSAgentQoSExtensionTestFramework, self).setUp()
self.config.set_override('extensions', ['qos'], 'agent')
self._set_pull_mock()
self.set_test_qos_rules(TEST_POLICY_ID1, [TEST_BW_LIMIT_RULE_1])
self.set_test_qos_rules(TEST_POLICY_ID2, [TEST_BW_LIMIT_RULE_2])
def _set_pull_mock(self):
self.qos_policies = {}
def _pull_mock(context, resource_type, resource_id):
return self.qos_policies[resource_id]
self.pull = mock.patch(
'neutron.api.rpc.handlers.resources_rpc.'
'ResourcesPullRpcApi.pull').start()
self.pull.side_effect = _pull_mock
def set_test_qos_rules(self, policy_id, policy_rules):
"""This function sets the policy test rules to be exposed."""
qos_policy = policy.QosPolicy(
context=None,
tenant_id=uuidutils.generate_uuid(),
id=policy_id,
name="Test Policy Name",
description="This is a policy for testing purposes",
shared=False,
rules=policy_rules)
qos_policy.obj_reset_changes()
self.qos_policies[policy_id] = qos_policy
def _create_test_port_dict(self, policy_id=None):
port_dict = super(OVSAgentQoSExtensionTestFramework,
self)._create_test_port_dict()
port_dict['qos_policy_id'] = policy_id
return port_dict
def _get_device_details(self, port, network):
dev = super(OVSAgentQoSExtensionTestFramework,
self)._get_device_details(port, network)
dev['qos_policy_id'] = port['qos_policy_id']
return dev
def _assert_bandwidth_limit_rule_is_set(self, port, rule):
max_rate, burst = (
self.agent.int_br.get_egress_bw_limit_for_port(port['vif_name']))
self.assertEqual(max_rate, rule.max_kbps)
self.assertEqual(burst, rule.max_burst_kbps)
def _assert_bandwidth_limit_rule_not_set(self, port):
max_rate, burst = (
self.agent.int_br.get_egress_bw_limit_for_port(port['vif_name']))
self.assertIsNone(max_rate)
self.assertIsNone(burst)
def wait_until_bandwidth_limit_rule_applied(self, port, rule):
l2_extensions.wait_until_bandwidth_limit_rule_applied(
self.agent.int_br, port['vif_name'], rule)
def _create_port_with_qos(self):
port_dict = self._create_test_port_dict()
port_dict['qos_policy_id'] = TEST_POLICY_ID1
self.setup_agent_and_ports([port_dict])
self.wait_until_ports_state(self.ports, up=True)
self.wait_until_bandwidth_limit_rule_applied(port_dict,
TEST_BW_LIMIT_RULE_1)
return port_dict
class TestOVSAgentQosExtension(OVSAgentQoSExtensionTestFramework):
def test_port_creation_with_bandwidth_limit(self):
"""Make sure bandwidth limit rules are set in low level to ports."""
self.setup_agent_and_ports(
port_dicts=self.create_test_ports(amount=1,
policy_id=TEST_POLICY_ID1))
self.wait_until_ports_state(self.ports, up=True)
for port in self.ports:
self._assert_bandwidth_limit_rule_is_set(
port, TEST_BW_LIMIT_RULE_1)
def test_port_creation_with_different_bandwidth_limits(self):
"""Make sure different types of policies end on the right ports."""
port_dicts = self.create_test_ports(amount=3)
port_dicts[0]['qos_policy_id'] = TEST_POLICY_ID1
port_dicts[1]['qos_policy_id'] = TEST_POLICY_ID2
self.setup_agent_and_ports(port_dicts)
self.wait_until_ports_state(self.ports, up=True)
self._assert_bandwidth_limit_rule_is_set(self.ports[0],
TEST_BW_LIMIT_RULE_1)
self._assert_bandwidth_limit_rule_is_set(self.ports[1],
TEST_BW_LIMIT_RULE_2)
self._assert_bandwidth_limit_rule_not_set(self.ports[2])
def test_simple_port_policy_update(self):
self.setup_agent_and_ports(
port_dicts=self.create_test_ports(amount=1,
policy_id=TEST_POLICY_ID1))
self.wait_until_ports_state(self.ports, up=True)
policy_copy = copy.deepcopy(self.qos_policies[TEST_POLICY_ID1])
policy_copy.rules[0].max_kbps = 500
policy_copy.rules[0].max_burst_kbps = 5
consumer_reg.push(resources.QOS_POLICY, policy_copy, events.UPDATED)
self.wait_until_bandwidth_limit_rule_applied(self.ports[0],
policy_copy.rules[0])
self._assert_bandwidth_limit_rule_is_set(self.ports[0],
policy_copy.rules[0])
def test_port_qos_disassociation(self):
"""Test that qos_policy_id set to None will remove all qos rules from
given port.
"""
port_dict = self._create_port_with_qos()
port_dict['qos_policy_id'] = None
self.agent.port_update(None, port=port_dict)
self.wait_until_bandwidth_limit_rule_applied(port_dict, None)
def test_port_qos_update_policy_id(self):
"""Test that change of qos policy id on given port refreshes all its
rules.
"""
port_dict = self._create_port_with_qos()
port_dict['qos_policy_id'] = TEST_POLICY_ID2
self.agent.port_update(None, port=port_dict)
self.wait_until_bandwidth_limit_rule_applied(port_dict,
TEST_BW_LIMIT_RULE_2)
def test_policy_rule_delete(self):
port_dict = self._create_port_with_qos()
policy_copy = copy.deepcopy(self.qos_policies[TEST_POLICY_ID1])
policy_copy.rules = list()
consumer_reg.push(resources.QOS_POLICY, policy_copy, events.UPDATED)
self.wait_until_bandwidth_limit_rule_applied(port_dict, None)

View File

@ -14,301 +14,34 @@
# License for the specific language governing permissions and limitations
# under the License.
import eventlet
import mock
import random
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import uuidutils
from neutron.agent.common import config as agent_config
from neutron.agent.common import ovs_lib
from neutron.agent.linux import interface
from neutron.agent.linux import polling
from neutron.agent.linux import utils as agent_utils
from neutron.common import config as common_config
from neutron.common import constants as n_const
from neutron.common import utils
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2.drivers.openvswitch.agent.common import config \
as ovs_config
from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_int
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_phys
from neutron.plugins.ml2.drivers.openvswitch.agent.openflow.ovs_ofctl \
import br_tun
from neutron.plugins.ml2.drivers.openvswitch.agent import ovs_neutron_agent \
as ovs_agent
from neutron.tests.functional.agent.linux import base
LOG = logging.getLogger(__name__)
from neutron.tests.functional.agent.l2 import base
class OVSAgentTestFramework(base.BaseOVSLinuxTestCase):
def setUp(self):
super(OVSAgentTestFramework, self).setUp()
agent_rpc = ('neutron.plugins.ml2.drivers.openvswitch.agent.'
'ovs_neutron_agent.OVSPluginApi')
mock.patch(agent_rpc).start()
mock.patch('neutron.agent.rpc.PluginReportStateAPI').start()
self.br_int = base.get_rand_name(n_const.DEVICE_NAME_MAX_LEN,
prefix='br-int')
self.br_tun = base.get_rand_name(n_const.DEVICE_NAME_MAX_LEN,
prefix='br-tun')
patch_name_len = n_const.DEVICE_NAME_MAX_LEN - len("-patch-tun")
self.patch_tun = "%s-patch-tun" % self.br_int[patch_name_len:]
self.patch_int = "%s-patch-int" % self.br_tun[patch_name_len:]
self.ovs = ovs_lib.BaseOVS()
self.config = self._configure_agent()
self.driver = interface.OVSInterfaceDriver(self.config)
def _get_config_opts(self):
config = cfg.ConfigOpts()
config.register_opts(common_config.core_opts)
config.register_opts(interface.OPTS)
config.register_opts(ovs_config.ovs_opts, "OVS")
config.register_opts(ovs_config.agent_opts, "AGENT")
agent_config.register_interface_driver_opts_helper(config)
agent_config.register_agent_state_opts_helper(config)
return config
def _configure_agent(self):
config = self._get_config_opts()
config.set_override(
'interface_driver',
'neutron.agent.linux.interface.OVSInterfaceDriver')
config.set_override('integration_bridge', self.br_int, "OVS")
config.set_override('ovs_integration_bridge', self.br_int)
config.set_override('tunnel_bridge', self.br_tun, "OVS")
config.set_override('int_peer_patch_port', self.patch_tun, "OVS")
config.set_override('tun_peer_patch_port', self.patch_int, "OVS")
config.set_override('host', 'ovs-agent')
return config
def _bridge_classes(self):
return {
'br_int': br_int.OVSIntegrationBridge,
'br_phys': br_phys.OVSPhysicalBridge,
'br_tun': br_tun.OVSTunnelBridge
}
def create_agent(self, create_tunnels=True):
if create_tunnels:
tunnel_types = [p_const.TYPE_VXLAN]
else:
tunnel_types = None
local_ip = '192.168.10.1'
bridge_mappings = {'physnet': self.br_int}
agent = ovs_agent.OVSNeutronAgent(self._bridge_classes(),
self.br_int, self.br_tun,
local_ip, bridge_mappings,
polling_interval=1,
tunnel_types=tunnel_types,
prevent_arp_spoofing=False,
conf=self.config)
self.addCleanup(self.ovs.delete_bridge, self.br_int)
if tunnel_types:
self.addCleanup(self.ovs.delete_bridge, self.br_tun)
agent.sg_agent = mock.Mock()
return agent
def start_agent(self, agent):
polling_manager = polling.InterfacePollingMinimizer()
self.addCleanup(polling_manager.stop)
polling_manager.start()
agent_utils.wait_until_true(
polling_manager._monitor.is_active)
agent.check_ovs_status = mock.Mock(
return_value=constants.OVS_NORMAL)
t = eventlet.spawn(agent.rpc_loop, polling_manager)
def stop_agent(agent, rpc_loop_thread):
agent.run_daemon_loop = False
rpc_loop_thread.wait()
self.addCleanup(stop_agent, agent, t)
def _bind_ports(self, ports, network, agent):
devices = []
for port in ports:
dev = OVSAgentTestFramework._get_device_details(port, network)
vif_name = port.get('vif_name')
vif_id = uuidutils.generate_uuid(),
vif_port = ovs_lib.VifPort(
vif_name, "%s" % vif_id, 'id-%s' % vif_id,
port.get('mac_address'), agent.int_br)
dev['vif_port'] = vif_port
devices.append(dev)
agent._bind_devices(devices)
def _create_test_port_dict(self):
return {'id': uuidutils.generate_uuid(),
'mac_address': utils.get_random_mac(
'fa:16:3e:00:00:00'.split(':')),
'fixed_ips': [{
'ip_address': '10.%d.%d.%d' % (
random.randint(3, 254),
random.randint(3, 254),
random.randint(3, 254))}],
'vif_name': base.get_rand_name(
self.driver.DEV_NAME_LEN, self.driver.DEV_NAME_PREFIX)}
def _create_test_network_dict(self):
return {'id': uuidutils.generate_uuid(),
'tenant_id': uuidutils.generate_uuid()}
def _plug_ports(self, network, ports, agent, ip_len=24):
for port in ports:
self.driver.plug(
network.get('id'), port.get('id'), port.get('vif_name'),
port.get('mac_address'),
agent.int_br.br_name, namespace=None)
ip_cidrs = ["%s/%s" % (port.get('fixed_ips')[0][
'ip_address'], ip_len)]
self.driver.init_l3(port.get('vif_name'), ip_cidrs, namespace=None)
@staticmethod
def _get_device_details(port, network):
dev = {'device': port['id'],
'port_id': port['id'],
'network_id': network['id'],
'network_type': 'vlan',
'physical_network': 'physnet',
'segmentation_id': 1,
'fixed_ips': port['fixed_ips'],
'device_owner': 'compute',
'admin_state_up': True}
return dev
def assert_bridge(self, br, exists=True):
self.assertEqual(exists, self.ovs.bridge_exists(br))
def assert_patch_ports(self, agent):
def get_peer(port):
return agent.int_br.db_get_val(
'Interface', port, 'options', check_error=True)
agent_utils.wait_until_true(
lambda: get_peer(self.patch_int) == {'peer': self.patch_tun})
agent_utils.wait_until_true(
lambda: get_peer(self.patch_tun) == {'peer': self.patch_int})
def assert_bridge_ports(self):
for port in [self.patch_tun, self.patch_int]:
self.assertTrue(self.ovs.port_exists(port))
def assert_vlan_tags(self, ports, agent):
for port in ports:
res = agent.int_br.db_get_val('Port', port.get('vif_name'), 'tag')
self.assertTrue(res)
class TestOVSAgent(OVSAgentTestFramework):
def _expected_plugin_rpc_call(self, call, expected_devices, is_up=True):
"""Helper to check expected rpc call are received
:param call: The call to check
:param expected_devices The device for which call is expected
:param is_up True if expected_devices are devices that are set up,
False if expected_devices are devices that are set down
"""
if is_up:
rpc_devices = [
dev for args in call.call_args_list for dev in args[0][1]]
else:
rpc_devices = [
dev for args in call.call_args_list for dev in args[0][2]]
return not (set(expected_devices) - set(rpc_devices))
def _create_ports(self, network, agent, trigger_resync=False):
ports = []
for x in range(3):
ports.append(self._create_test_port_dict())
def mock_device_raise_exception(context, devices_up, devices_down,
agent_id, host=None):
agent.plugin_rpc.update_device_list.side_effect = (
mock_update_device)
raise Exception('Exception to trigger resync')
def mock_device_details(context, devices, agent_id, host=None):
details = []
for port in ports:
if port['id'] in devices:
dev = OVSAgentTestFramework._get_device_details(
port, network)
details.append(dev)
return {'devices': details, 'failed_devices': []}
def mock_update_device(context, devices_up, devices_down, agent_id,
host=None):
dev_up = []
dev_down = []
for port in ports:
if devices_up and port['id'] in devices_up:
dev_up.append(port['id'])
if devices_down and port['id'] in devices_down:
dev_down.append({'device': port['id'], 'exists': True})
return {'devices_up': dev_up,
'failed_devices_up': [],
'devices_down': dev_down,
'failed_devices_down': []}
(agent.plugin_rpc.get_devices_details_list_and_failed_devices.
side_effect) = mock_device_details
if trigger_resync:
agent.plugin_rpc.update_device_list.side_effect = (
mock_device_raise_exception)
else:
agent.plugin_rpc.update_device_list.side_effect = (
mock_update_device)
return ports
class TestOVSAgent(base.OVSAgentTestFramework):
def test_port_creation_and_deletion(self):
agent = self.create_agent()
self.start_agent(agent)
network = self._create_test_network_dict()
ports = self._create_ports(network, agent)
self._plug_ports(network, ports, agent)
up_ports_ids = [p['id'] for p in ports]
agent_utils.wait_until_true(
lambda: self._expected_plugin_rpc_call(
agent.plugin_rpc.update_device_list, up_ports_ids))
down_ports_ids = [p['id'] for p in ports]
for port in ports:
agent.int_br.delete_port(port['vif_name'])
agent_utils.wait_until_true(
lambda: self._expected_plugin_rpc_call(
agent.plugin_rpc.update_device_list, down_ports_ids, False))
self.setup_agent_and_ports(
port_dicts=self.create_test_ports())
self.wait_until_ports_state(self.ports, up=True)
for port in self.ports:
self.agent.int_br.delete_port(port['vif_name'])
self.wait_until_ports_state(self.ports, up=False)
def test_resync_devices_set_up_after_exception(self):
agent = self.create_agent()
self.start_agent(agent)
network = self._create_test_network_dict()
ports = self._create_ports(network, agent, True)
self._plug_ports(network, ports, agent)
ports_ids = [p['id'] for p in ports]
agent_utils.wait_until_true(
lambda: self._expected_plugin_rpc_call(
agent.plugin_rpc.update_device_list, ports_ids))
self.setup_agent_and_ports(
port_dicts=self.create_test_ports(),
trigger_resync=True)
self.wait_until_ports_state(self.ports, up=True)
def test_port_vlan_tags(self):
agent = self.create_agent()
self.start_agent(agent)
network = self._create_test_network_dict()
ports = self._create_ports(network, agent)
ports_ids = [p['id'] for p in ports]
self._plug_ports(network, ports, agent)
agent_utils.wait_until_true(
lambda: self._expected_plugin_rpc_call(
agent.plugin_rpc.update_device_list, ports_ids))
self.assert_vlan_tags(ports, agent)
self.setup_agent_and_ports(
port_dicts=self.create_test_ports(),
trigger_resync=True)
self.wait_until_ports_state(self.ports, up=True)
self.assert_vlan_tags(self.ports, self.agent)
def test_assert_bridges_ports_vxlan(self):
agent = self.create_agent()

View File

@ -311,6 +311,17 @@ class OVSBridgeTestCase(OVSBridgeTestBase):
controller,
'connection_mode'))
def test_egress_bw_limit(self):
port_name, _ = self.create_ovs_port()
self.br.create_egress_bw_limit_for_port(port_name, 700, 70)
max_rate, burst = self.br.get_egress_bw_limit_for_port(port_name)
self.assertEqual(700, max_rate)
self.assertEqual(70, burst)
self.br.delete_egress_bw_limit_for_port(port_name)
max_rate, burst = self.br.get_egress_bw_limit_for_port(port_name)
self.assertIsNone(max_rate)
self.assertIsNone(burst)
class OVSLibTestCase(base.BaseOVSLinuxTestCase):

View File

@ -11,6 +11,8 @@
# under the License.
import time
import urllib
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse
@ -65,7 +67,10 @@ 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',
'rule_types': 'qos',
}
service_prefix = service_resource_prefix_map.get(
plural_name)
@ -90,7 +95,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 +626,88 @@ 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, **filters):
if filters:
uri = '%s/qos/policies?%s' % (self.uri_prefix,
urllib.urlencode(filters))
else:
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, tenant_id=None):
uri = '%s/qos/policies' % self.uri_prefix
post_data = {'policy': {
'name': name,
'description': description,
'shared': shared
}}
if tenant_id is not None:
post_data['policy']['tenant_id'] = tenant_id
resp, body = self.post(uri, self.serialize(post_data))
body = self.deserialize_single(body)
self.expected_success(201, resp.status)
return service_client.ResponseBody(resp, body)
def update_qos_policy(self, policy_id, **kwargs):
uri = '%s/qos/policies/%s' % (self.uri_prefix, policy_id)
post_data = self.serialize({'policy': kwargs})
resp, body = self.put(uri, post_data)
body = self.deserialize_single(body)
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)
post_data = self.serialize(
{'bandwidth_limit_rule': {
'max_kbps': max_kbps,
'max_burst_kbps': max_burst_kbps}
})
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, **kwargs):
uri = '%s/qos/policies/%s/bandwidth_limit_rules/%s' % (
self.uri_prefix, policy_id, rule_id)
post_data = {'bandwidth_limit_rule': kwargs}
resp, body = self.put(uri, json.dumps(post_data))
body = self.deserialize_single(body)
self.expected_success(200, resp.status)
return service_client.ResponseBody(resp, body)
def delete_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.delete(uri)
self.expected_success(204, resp.status)
return service_client.ResponseBody(resp, body)
def list_qos_rule_types(self):
uri = '%s/qos/rule-types' % self.uri_prefix
resp, body = self.get(uri)
self.expected_success(200, resp.status)
body = json.loads(body)
return service_client.ResponseBody(resp, body)

View File

View File

@ -0,0 +1,52 @@
# 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_config import cfg
from neutron.agent.l2.extensions import manager as ext_manager
from neutron.tests import base
class TestAgentExtensionsManager(base.BaseTestCase):
def setUp(self):
super(TestAgentExtensionsManager, self).setUp()
mock.patch('neutron.agent.l2.extensions.qos.QosAgentExtension',
autospec=True).start()
conf = cfg.CONF
ext_manager.register_opts(conf)
cfg.CONF.set_override('extensions', ['qos'], 'agent')
self.manager = ext_manager.AgentExtensionsManager(conf)
def _get_extension(self):
return self.manager.extensions[0].obj
def test_initialize(self):
connection = object()
self.manager.initialize(connection, 'fake_driver_type')
ext = self._get_extension()
ext.initialize.assert_called_once_with(connection, 'fake_driver_type')
def test_handle_port(self):
context = object()
data = object()
self.manager.handle_port(context, data)
ext = self._get_extension()
ext.handle_port.assert_called_once_with(context, data)
def test_delete_port(self):
context = object()
data = object()
self.manager.delete_port(context, data)
ext = self._get_extension()
ext.delete_port.assert_called_once_with(context, data)

View File

@ -0,0 +1,187 @@
# 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
from neutron.api.rpc.callbacks.consumer import registry
from neutron.api.rpc.callbacks import events
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron import context
from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
from neutron.tests import base
TEST_POLICY = object()
class QosExtensionBaseTestCase(base.BaseTestCase):
def setUp(self):
super(QosExtensionBaseTestCase, self).setUp()
self.qos_ext = qos.QosAgentExtension()
self.context = context.get_admin_context()
self.connection = mock.Mock()
# Don't rely on used driver
mock.patch(
'neutron.manager.NeutronManager.load_class_for_provider',
return_value=lambda: mock.Mock(spec=qos.QosAgentDriver)
).start()
class QosExtensionRpcTestCase(QosExtensionBaseTestCase):
def setUp(self):
super(QosExtensionRpcTestCase, self).setUp()
self.qos_ext.initialize(
self.connection, constants.EXTENSION_DRIVER_TYPE)
self.pull_mock = mock.patch.object(
self.qos_ext.resource_rpc, 'pull',
return_value=TEST_POLICY).start()
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_ext._process_reset_port = mock.Mock()
self.qos_ext.handle_port(self.context, port)
self.qos_ext._process_reset_port.assert_called_with(port)
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_ext.handle_port(self.context, port)
# we make sure the underlaying qos driver is called with the
# right parameters
self.qos_ext.qos_driver.create.assert_called_once_with(
port, TEST_POLICY)
self.assertEqual(port,
self.qos_ext.qos_policy_ports[qos_policy_id][port_id])
self.assertTrue(port_id in self.qos_ext.known_ports)
def test_handle_known_port(self):
port_obj1 = self._create_test_port_dict()
port_obj2 = dict(port_obj1)
self.qos_ext.handle_port(self.context, port_obj1)
self.qos_ext.qos_driver.reset_mock()
self.qos_ext.handle_port(self.context, port_obj2)
self.assertFalse(self.qos_ext.qos_driver.create.called)
def test_handle_known_port_change_policy_id(self):
port = self._create_test_port_dict()
self.qos_ext.handle_port(self.context, port)
self.qos_ext.resource_rpc.pull.reset_mock()
port['qos_policy_id'] = uuidutils.generate_uuid()
self.qos_ext.handle_port(self.context, port)
self.pull_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
def test_delete_known_port(self):
port = self._create_test_port_dict()
port_id = port['port_id']
self.qos_ext.handle_port(self.context, port)
self.qos_ext.qos_driver.reset_mock()
self.qos_ext.delete_port(self.context, port)
self.qos_ext.qos_driver.delete.assert_called_with(port, None)
self.assertNotIn(port_id, self.qos_ext.known_ports)
def test_delete_unknown_port(self):
port = self._create_test_port_dict()
port_id = port['port_id']
self.qos_ext.delete_port(self.context, port)
self.assertFalse(self.qos_ext.qos_driver.delete.called)
self.assertNotIn(port_id, self.qos_ext.known_ports)
def test__handle_notification_ignores_all_event_types_except_updated(self):
with mock.patch.object(
self.qos_ext, '_process_update_policy') as update_mock:
for event_type in set(events.VALID) - {events.UPDATED}:
self.qos_ext._handle_notification(object(), event_type)
self.assertFalse(update_mock.called)
def test__handle_notification_passes_update_events(self):
with mock.patch.object(
self.qos_ext, '_process_update_policy') as update_mock:
policy = mock.Mock()
self.qos_ext._handle_notification(policy, events.UPDATED)
update_mock.assert_called_with(policy)
def test__process_update_policy(self):
port1 = self._create_test_port_dict()
port2 = self._create_test_port_dict()
self.qos_ext.qos_policy_ports = {
port1['qos_policy_id']: {port1['port_id']: port1},
port2['qos_policy_id']: {port2['port_id']: port2},
}
policy = mock.Mock()
policy.id = port1['qos_policy_id']
self.qos_ext._process_update_policy(policy)
self.qos_ext.qos_driver.update.assert_called_with(port1, policy)
self.qos_ext.qos_driver.update.reset_mock()
policy.id = port2['qos_policy_id']
self.qos_ext._process_update_policy(policy)
self.qos_ext.qos_driver.update.assert_called_with(port2, policy)
def test__process_reset_port(self):
port1 = self._create_test_port_dict()
port2 = self._create_test_port_dict()
port1_id = port1['port_id']
port2_id = port2['port_id']
self.qos_ext.qos_policy_ports = {
port1['qos_policy_id']: {port1_id: port1},
port2['qos_policy_id']: {port2_id: port2},
}
self.qos_ext.known_ports = {port1_id, port2_id}
self.qos_ext._process_reset_port(port1)
self.qos_ext.qos_driver.delete.assert_called_with(port1, None)
self.assertNotIn(port1_id, self.qos_ext.known_ports)
self.assertIn(port2_id, self.qos_ext.known_ports)
self.qos_ext.qos_driver.delete.reset_mock()
self.qos_ext._process_reset_port(port2)
self.qos_ext.qos_driver.delete.assert_called_with(port2, None)
self.assertNotIn(port2_id, self.qos_ext.known_ports)
class QosExtensionInitializeTestCase(QosExtensionBaseTestCase):
@mock.patch.object(registry, 'subscribe')
@mock.patch.object(resources_rpc, 'ResourcesPushRpcCallback')
def test_initialize_subscribed_to_rpc(self, rpc_mock, subscribe_mock):
self.qos_ext.initialize(
self.connection, constants.EXTENSION_DRIVER_TYPE)
self.connection.create_consumer.assert_has_calls(
[mock.call(
resources_rpc.resource_type_versioned_topic(resource_type),
[rpc_mock()],
fanout=True)
for resource_type in self.qos_ext.SUPPORTED_RESOURCES]
)
subscribe_mock.assert_called_with(mock.ANY, resources.QOS_POLICY)

View File

@ -0,0 +1,56 @@
# 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.api.rpc.callbacks.consumer import registry
from neutron.tests import base
class ConsumerRegistryTestCase(base.BaseTestCase):
def setUp(self):
super(ConsumerRegistryTestCase, self).setUp()
def test__get_manager_is_singleton(self):
self.assertIs(registry._get_manager(), registry._get_manager())
@mock.patch.object(registry, '_get_manager')
def test_subscribe(self, manager_mock):
callback = lambda: None
registry.subscribe(callback, 'TYPE')
manager_mock().register.assert_called_with(callback, 'TYPE')
@mock.patch.object(registry, '_get_manager')
def test_unsubscribe(self, manager_mock):
callback = lambda: None
registry.unsubscribe(callback, 'TYPE')
manager_mock().unregister.assert_called_with(callback, 'TYPE')
@mock.patch.object(registry, '_get_manager')
def test_clear(self, manager_mock):
registry.clear()
manager_mock().clear.assert_called_with()
@mock.patch.object(registry, '_get_manager')
def test_push(self, manager_mock):
resource_type_ = object()
resource_ = object()
event_type_ = object()
callback1 = mock.Mock()
callback2 = mock.Mock()
callbacks = {callback1, callback2}
manager_mock().get_callbacks.return_value = callbacks
registry.push(resource_type_, resource_, event_type_)
for callback in callbacks:
callback.assert_called_with(resource_, event_type_)

View File

@ -0,0 +1,81 @@
# 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 exceptions
from neutron.api.rpc.callbacks.producer import registry
from neutron.api.rpc.callbacks import resources
from neutron.objects.qos import policy
from neutron.tests.unit.services.qos import base
class ProducerRegistryTestCase(base.BaseQosTestCase):
def test_pull_returns_callback_result(self):
policy_obj = policy.QosPolicy(context=None)
def _fake_policy_cb(*args, **kwargs):
return policy_obj
registry.provide(_fake_policy_cb, resources.QOS_POLICY)
self.assertEqual(
policy_obj,
registry.pull(resources.QOS_POLICY, 'fake_id'))
def test_pull_does_not_raise_on_none(self):
def _none_cb(*args, **kwargs):
pass
registry.provide(_none_cb, resources.QOS_POLICY)
obj = registry.pull(resources.QOS_POLICY, 'fake_id')
self.assertIsNone(obj)
def test_pull_raises_on_wrong_object_type(self):
def _wrong_type_cb(*args, **kwargs):
return object()
registry.provide(_wrong_type_cb, resources.QOS_POLICY)
self.assertRaises(
exceptions.CallbackWrongResourceType,
registry.pull, resources.QOS_POLICY, 'fake_id')
def test_pull_raises_on_callback_not_found(self):
self.assertRaises(
exceptions.CallbackNotFound,
registry.pull, resources.QOS_POLICY, 'fake_id')
def test__get_manager_is_singleton(self):
self.assertIs(registry._get_manager(), registry._get_manager())
def test_unprovide(self):
def _fake_policy_cb(*args, **kwargs):
pass
registry.provide(_fake_policy_cb, resources.QOS_POLICY)
registry.unprovide(_fake_policy_cb, resources.QOS_POLICY)
self.assertRaises(
exceptions.CallbackNotFound,
registry.pull, resources.QOS_POLICY, 'fake_id')
def test_clear_unprovides_all_producers(self):
def _fake_policy_cb(*args, **kwargs):
pass
registry.provide(_fake_policy_cb, resources.QOS_POLICY)
registry.clear()
self.assertRaises(
exceptions.CallbackNotFound,
registry.pull, resources.QOS_POLICY, 'fake_id')

View File

@ -0,0 +1,140 @@
# 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.api.rpc.callbacks import exceptions as rpc_exc
from neutron.api.rpc.callbacks import resource_manager
from neutron.callbacks import exceptions as exceptions
from neutron.tests.unit.services.qos import base
IS_VALID_RESOURCE_TYPE = (
'neutron.api.rpc.callbacks.resources.is_valid_resource_type')
class ResourceCallbacksManagerTestCaseMixin(object):
def test_register_fails_on_invalid_type(self):
self.assertRaises(
exceptions.Invalid,
self.mgr.register, lambda: None, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_clear_unregisters_all_callbacks(self, *mocks):
self.mgr.register(lambda: None, 'TYPE1')
self.mgr.register(lambda: None, 'TYPE2')
self.mgr.clear()
self.assertEqual([], self.mgr.get_subscribed_types())
def test_unregister_fails_on_invalid_type(self):
self.assertRaises(
exceptions.Invalid,
self.mgr.unregister, lambda: None, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_unregister_fails_on_unregistered_callback(self, *mocks):
self.assertRaises(
rpc_exc.CallbackNotFound,
self.mgr.unregister, lambda: None, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_unregister_unregisters_callback(self, *mocks):
callback = lambda: None
self.mgr.register(callback, 'TYPE')
self.mgr.unregister(callback, 'TYPE')
self.assertEqual([], self.mgr.get_subscribed_types())
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test___init___does_not_reset_callbacks(self, *mocks):
callback = lambda: None
self.mgr.register(callback, 'TYPE')
resource_manager.ProducerResourceCallbacksManager()
self.assertEqual(['TYPE'], self.mgr.get_subscribed_types())
class ProducerResourceCallbacksManagerTestCase(
base.BaseQosTestCase, ResourceCallbacksManagerTestCaseMixin):
def setUp(self):
super(ProducerResourceCallbacksManagerTestCase, self).setUp()
self.mgr = self.prod_mgr
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_register_registers_callback(self, *mocks):
callback = lambda: None
self.mgr.register(callback, 'TYPE')
self.assertEqual(callback, self.mgr.get_callback('TYPE'))
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_register_fails_on_multiple_calls(self, *mocks):
self.mgr.register(lambda: None, 'TYPE')
self.assertRaises(
rpc_exc.CallbacksMaxLimitReached,
self.mgr.register, lambda: None, 'TYPE')
def test_get_callback_fails_on_invalid_type(self):
self.assertRaises(
exceptions.Invalid,
self.mgr.get_callback, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_get_callback_fails_on_unregistered_callback(
self, *mocks):
self.assertRaises(
rpc_exc.CallbackNotFound,
self.mgr.get_callback, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_get_callback_returns_proper_callback(self, *mocks):
callback1 = lambda: None
callback2 = lambda: None
self.mgr.register(callback1, 'TYPE1')
self.mgr.register(callback2, 'TYPE2')
self.assertEqual(callback1, self.mgr.get_callback('TYPE1'))
self.assertEqual(callback2, self.mgr.get_callback('TYPE2'))
class ConsumerResourceCallbacksManagerTestCase(
base.BaseQosTestCase, ResourceCallbacksManagerTestCaseMixin):
def setUp(self):
super(ConsumerResourceCallbacksManagerTestCase, self).setUp()
self.mgr = self.cons_mgr
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_register_registers_callback(self, *mocks):
callback = lambda: None
self.mgr.register(callback, 'TYPE')
self.assertEqual({callback}, self.mgr.get_callbacks('TYPE'))
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_register_succeeds_on_multiple_calls(self, *mocks):
callback1 = lambda: None
callback2 = lambda: None
self.mgr.register(callback1, 'TYPE')
self.mgr.register(callback2, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_get_callbacks_fails_on_unregistered_callback(
self, *mocks):
self.assertRaises(
rpc_exc.CallbackNotFound,
self.mgr.get_callbacks, 'TYPE')
@mock.patch(IS_VALID_RESOURCE_TYPE, return_value=True)
def test_get_callbacks_returns_proper_callbacks(self, *mocks):
callback1 = lambda: None
callback2 = lambda: None
self.mgr.register(callback1, 'TYPE1')
self.mgr.register(callback2, 'TYPE2')
self.assertEqual(set([callback1]), self.mgr.get_callbacks('TYPE1'))
self.assertEqual(set([callback2]), self.mgr.get_callbacks('TYPE2'))

View File

@ -0,0 +1,54 @@
# 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 resources
from neutron.objects.qos import policy
from neutron.tests import base
class GetResourceTypeTestCase(base.BaseTestCase):
def test_get_resource_type_none(self):
self.assertIsNone(resources.get_resource_type(None))
def test_get_resource_type_wrong_type(self):
self.assertIsNone(resources.get_resource_type(object()))
def test_get_resource_type(self):
# we could use any other registered NeutronObject type here
self.assertEqual(policy.QosPolicy.obj_name(),
resources.get_resource_type(policy.QosPolicy()))
class IsValidResourceTypeTestCase(base.BaseTestCase):
def test_known_type(self):
# it could be any other NeutronObject, assuming it's known to RPC
# callbacks
self.assertTrue(resources.is_valid_resource_type(
policy.QosPolicy.obj_name()))
def test_unknown_type(self):
self.assertFalse(
resources.is_valid_resource_type('unknown-resource-type'))
class GetResourceClsTestCase(base.BaseTestCase):
def test_known_type(self):
# it could be any other NeutronObject, assuming it's known to RPC
# callbacks
self.assertEqual(policy.QosPolicy,
resources.get_resource_cls(resources.QOS_POLICY))
def test_unknown_type(self):
self.assertIsNone(resources.get_resource_cls('unknown-resource-type'))

View File

@ -0,0 +1,222 @@
# Copyright (c) 2015 Mellanox Technologies, Ltd
#
# 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_versionedobjects import base as obj_base
from oslo_versionedobjects import fields as obj_fields
import testtools
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron.common import topics
from neutron import context
from neutron.objects import base as objects_base
from neutron.tests import base
def _create_test_dict():
return {'id': 'uuid',
'field': 'foo'}
def _create_test_resource(context=None):
resource_dict = _create_test_dict()
resource = FakeResource(context, **resource_dict)
resource.obj_reset_changes()
return resource
@obj_base.VersionedObjectRegistry.register
class FakeResource(objects_base.NeutronObject):
fields = {
'id': obj_fields.UUIDField(),
'field': obj_fields.StringField()
}
@classmethod
def get_objects(cls, context, **kwargs):
return list()
class ResourcesRpcBaseTestCase(base.BaseTestCase):
def setUp(self):
super(ResourcesRpcBaseTestCase, self).setUp()
self.context = context.get_admin_context()
class _ValidateResourceTypeTestCase(base.BaseTestCase):
def setUp(self):
super(_ValidateResourceTypeTestCase, self).setUp()
self.is_valid_mock = mock.patch.object(
resources_rpc.resources, 'is_valid_resource_type').start()
def test_valid_type(self):
self.is_valid_mock.return_value = True
resources_rpc._validate_resource_type('foo')
def test_invalid_type(self):
self.is_valid_mock.return_value = False
with testtools.ExpectedException(
resources_rpc.InvalidResourceTypeClass):
resources_rpc._validate_resource_type('foo')
class _ResourceTypeVersionedTopicTestCase(base.BaseTestCase):
@mock.patch.object(resources_rpc, '_validate_resource_type')
def test_resource_type_versioned_topic(self, validate_mock):
obj_name = FakeResource.obj_name()
expected = topics.RESOURCE_TOPIC_PATTERN % {
'resource_type': 'FakeResource', 'version': '1.0'}
with mock.patch.object(resources_rpc.resources, 'get_resource_cls',
return_value=FakeResource):
observed = resources_rpc.resource_type_versioned_topic(obj_name)
self.assertEqual(expected, observed)
class ResourcesPullRpcApiTestCase(ResourcesRpcBaseTestCase):
def setUp(self):
super(ResourcesPullRpcApiTestCase, self).setUp()
mock.patch.object(resources_rpc, '_validate_resource_type').start()
mock.patch('neutron.api.rpc.callbacks.resources.get_resource_cls',
return_value=FakeResource).start()
self.rpc = resources_rpc.ResourcesPullRpcApi()
mock.patch.object(self.rpc, 'client').start()
self.cctxt_mock = self.rpc.client.prepare.return_value
def test_is_singleton(self):
self.assertIs(self.rpc, resources_rpc.ResourcesPullRpcApi())
def test_pull(self):
expected_obj = _create_test_resource(self.context)
resource_id = expected_obj.id
self.cctxt_mock.call.return_value = expected_obj.obj_to_primitive()
result = self.rpc.pull(
self.context, FakeResource.obj_name(), resource_id)
self.cctxt_mock.call.assert_called_once_with(
self.context, 'pull', resource_type='FakeResource',
version=FakeResource.VERSION, resource_id=resource_id)
self.assertEqual(expected_obj, result)
def test_pull_resource_not_found(self):
resource_dict = _create_test_dict()
resource_id = resource_dict['id']
self.cctxt_mock.call.return_value = None
with testtools.ExpectedException(resources_rpc.ResourceNotFound):
self.rpc.pull(self.context, FakeResource.obj_name(),
resource_id)
class ResourcesPullRpcCallbackTestCase(ResourcesRpcBaseTestCase):
def setUp(self):
super(ResourcesPullRpcCallbackTestCase, self).setUp()
self.callbacks = resources_rpc.ResourcesPullRpcCallback()
self.resource_obj = _create_test_resource(self.context)
def test_pull(self):
resource_dict = _create_test_dict()
with mock.patch.object(
resources_rpc.prod_registry, 'pull',
return_value=self.resource_obj) as registry_mock:
primitive = self.callbacks.pull(
self.context, resource_type=FakeResource.obj_name(),
version=FakeResource.VERSION,
resource_id=self.resource_obj.id)
registry_mock.assert_called_once_with(
'FakeResource', self.resource_obj.id, context=self.context)
self.assertEqual(resource_dict,
primitive['versioned_object.data'])
self.assertEqual(self.resource_obj.obj_to_primitive(), primitive)
@mock.patch.object(FakeResource, 'obj_to_primitive')
def test_pull_no_backport_for_latest_version(self, to_prim_mock):
with mock.patch.object(resources_rpc.prod_registry, 'pull',
return_value=self.resource_obj):
self.callbacks.pull(
self.context, resource_type=FakeResource.obj_name(),
version=FakeResource.VERSION,
resource_id=self.resource_obj.id)
to_prim_mock.assert_called_with(target_version=None)
@mock.patch.object(FakeResource, 'obj_to_primitive')
def test_pull_backports_to_older_version(self, to_prim_mock):
with mock.patch.object(resources_rpc.prod_registry, 'pull',
return_value=self.resource_obj):
self.callbacks.pull(
self.context, resource_type=FakeResource.obj_name(),
version='0.9', # less than initial version 1.0
resource_id=self.resource_obj.id)
to_prim_mock.assert_called_with(target_version='0.9')
class ResourcesPushRpcApiTestCase(ResourcesRpcBaseTestCase):
def setUp(self):
super(ResourcesPushRpcApiTestCase, self).setUp()
mock.patch.object(resources_rpc.n_rpc, 'get_client').start()
mock.patch.object(resources_rpc, '_validate_resource_type').start()
self.rpc = resources_rpc.ResourcesPushRpcApi()
self.cctxt_mock = self.rpc.client.prepare.return_value
self.resource_obj = _create_test_resource(self.context)
def test__prepare_object_fanout_context(self):
expected_topic = topics.RESOURCE_TOPIC_PATTERN % {
'resource_type': resources.get_resource_type(self.resource_obj),
'version': self.resource_obj.VERSION}
with mock.patch.object(resources_rpc.resources, 'get_resource_cls',
return_value=FakeResource):
observed = self.rpc._prepare_object_fanout_context(
self.resource_obj)
self.rpc.client.prepare.assert_called_once_with(
fanout=True, topic=expected_topic)
self.assertEqual(self.cctxt_mock, observed)
def test_pushy(self):
with mock.patch.object(resources_rpc.resources, 'get_resource_cls',
return_value=FakeResource):
self.rpc.push(
self.context, self.resource_obj, 'TYPE')
self.cctxt_mock.cast.assert_called_once_with(
self.context, 'push',
resource=self.resource_obj.obj_to_primitive(),
event_type='TYPE')
class ResourcesPushRpcCallbackTestCase(ResourcesRpcBaseTestCase):
def setUp(self):
super(ResourcesPushRpcCallbackTestCase, self).setUp()
mock.patch.object(resources_rpc, '_validate_resource_type').start()
mock.patch.object(
resources_rpc.resources,
'get_resource_cls', return_value=FakeResource).start()
self.resource_obj = _create_test_resource(self.context)
self.resource_prim = self.resource_obj.obj_to_primitive()
self.callbacks = resources_rpc.ResourcesPushRpcCallback()
@mock.patch.object(resources_rpc.cons_registry, 'push')
def test_push(self, reg_push_mock):
self.callbacks.push(self.context, self.resource_prim, 'TYPE')
reg_push_mock.assert_called_once_with(self.resource_obj.obj_name(),
self.resource_obj, 'TYPE')

View File

@ -679,3 +679,24 @@ 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))
class TestRoundVal(base.BaseTestCase):
def test_round_val_ok(self):
for expected, value in ((0, 0),
(0, 0.1),
(1, 0.5),
(1, 1.49),
(2, 1.5)):
self.assertEqual(expected, utils.round_val(value))

Some files were not shown because too many files have changed in this diff Show More