Merge "QoS implementation(Part1: Qos Plugin)"

devstack/plugin.sh

@@ -291,6 +291,11 @@ function start_central_neutron_server {
         iniset $NEUTRON_CONF.$server_index sfc drivers tricircle_sfc
         iniset $NEUTRON_CONF.$server_index flowclassifier drivers tricircle_fc
     fi
+
+    if [ "$TRICIRCLE_ENABLE_QOS" == "True" ]; then
+        service_plugins+=",tricircle.network.central_qos_plugin.TricircleQosPlugin"
+    fi
+
     if [ -n service_plugins ]; then
         service_plugins=$(echo $service_plugins| sed 's/^,//')
         iniset $NEUTRON_CONF.$server_index DEFAULT service_plugins "$service_plugins"
@@ -324,6 +329,17 @@ function start_central_neutron_server {
     iniset $NEUTRON_CONF.$server_index tricircle enable_api_gateway False
     # default value of bridge_network_type is vxlan
 
+    if [ "$TRICIRCLE_ENABLE_QOS" == "True" ]; then
+        local p_exist=$(grep "^extension_drivers" /$Q_PLUGIN_CONF_FILE)
+        if [[ $p_exist != "" ]];then
+            if ! [[ $(echo $p_exist | grep "qos") ]];then
+                sed -i "s/$p_exist/$p_exist,qos/g" /$Q_PLUGIN_CONF_FILE
+            fi
+        else
+            sed -i "s/^\[ml2\]/\[ml2\]\nextension_drivers = qos/g" /$Q_PLUGIN_CONF_FILE
+        fi
+    fi
+
     recreate_database $Q_DB_NAME$server_index
     $NEUTRON_BIN_DIR/neutron-db-manage --config-file $NEUTRON_CONF.$server_index --config-file /$Q_PLUGIN_CONF_FILE upgrade head
 

devstack/settings

@@ -13,6 +13,7 @@ TRICIRCLE_DEPLOY_WITH_CELL=${TRICIRCLE_DEPLOY_WITH_CELL:-False}
 # extensions working with tricircle
 TRICIRCLE_ENABLE_TRUNK=${TRICIRCLE_ENABLE_TRUNK:-False}
 TRICIRCLE_ENABLE_SFC=${TRICIRCLE_ENABLE_SFC:-False}
+TRICIRCLE_ENABLE_QOS=${TRICIRCLE_ENABLE_QOS:-False}
 
 # these default settings are used for devstack based gate/check jobs
 TRICIRCLE_DEFAULT_VLAN_BRIDGE=${TRICIRCLE_DEFAULT_VLAN_BRIDGE:-br-vlan}

releasenotes/notes/add-qos-policy-rule-f8f1529d7ad5d888.yaml

@@ -0,0 +1,5 @@
+---
+features:
+  - Provide central Neutron QoS plugin and implement QoS driver. Support QoS
+    policy creation, update and delete, QoS policy binding with network or
+    port.

setup.cfg

@@ -64,6 +64,8 @@ tricircle.network.type_drivers =
     vlan = tricircle.network.drivers.type_vlan:VLANTypeDriver
     vxlan = tricircle.network.drivers.type_vxlan:VxLANTypeDriver
     flat = tricircle.network.drivers.type_flat:FlatTypeDriver
+tricircle.network.extension_drivers =
+    qos = neutron.plugins.ml2.extensions.qos:QosExtensionDriver
 networking_sfc.flowclassifier.drivers =
     tricircle_fc = tricircle.network.central_fc_driver:TricircleFcDriver
 networking_sfc.sfc.drivers =

tricircle/common/client.py

@@ -209,6 +209,12 @@ class Client(object):
                     if (handle_obj.support_resource[resource] & index) == 0:
                         continue
                     self.operation_resources_map[operation].add(resource)
+                    if resource == 'qos_policy':
+                        setattr(self, '%s_qos_policies' % operation,
+                                functools.partial(
+                                    getattr(self, '%s_resources' % operation),
+                                    resource))
+                    else:
                         setattr(self, '%s_%ss' % (operation, resource),
                                 functools.partial(
                                 getattr(self, '%s_resources' % operation),

tricircle/common/constants.py

@@ -36,6 +36,7 @@ RT_ROUTER = 'router'
 RT_NS_ROUTER = 'ns_router'
 RT_SG = 'security_group'
 RT_FIP = 'floatingip'
+RT_QOS = 'qos_policy'
 
 REAL_SHADOW_TYPE_MAP = {
     RT_NETWORK: RT_SD_NETWORK,
@@ -48,7 +49,7 @@ REAL_SHADOW_TYPE_MAP = {
 def is_valid_resource_type(resource_type):
     resource_type_table = [RT_NETWORK, RT_SUBNET, RT_PORT, RT_ROUTER, RT_SG,
                            RT_TRUNK, RT_PORT_PAIR, RT_PORT_PAIR_GROUP,
-                           RT_FLOW_CLASSIFIER, RT_PORT_CHAIN]
+                           RT_FLOW_CLASSIFIER, RT_PORT_CHAIN, RT_QOS]
     return resource_type in resource_type_table
 
 
@@ -113,6 +114,10 @@ JT_SHADOW_PORT_SETUP = 'shadow_port_setup'
 JT_TRUNK_SYNC = 'trunk_sync'
 JT_SFC_SYNC = 'sfc_sync'
 JT_RESOURCE_RECYCLE = 'resource_recycle'
+JT_QOS_CREATE = 'qos_create'
+JT_QOS_UPDATE = 'qos_update'
+JT_QOS_DELETE = 'qos_delete'
+JT_SYNC_QOS_RULE = 'sync_qos_rule'
 
 # network type
 NT_LOCAL = 'local'
@@ -148,7 +153,17 @@ job_resource_map = {
     JT_SFC_SYNC: [(None, "pod_id"),
                   (RT_PORT_CHAIN, "portchain_id"),
                   (RT_NETWORK, "network_id")],
-    JT_RESOURCE_RECYCLE: [(None, "project_id")]
+    JT_RESOURCE_RECYCLE: [(None, "project_id")],
+    JT_QOS_CREATE: [(None, "pod_id"),
+                    (RT_QOS, "policy_id"),
+                    (None, "res_type"),
+                    (None, "res_id")],
+    JT_QOS_UPDATE: [(None, "pod_id"),
+                    (RT_QOS, "policy_id")],
+    JT_QOS_DELETE: [(None, "pod_id"),
+                    (RT_QOS, "policy_id")],
+    JT_SYNC_QOS_RULE: [(None, "rule_id"),
+                       (RT_QOS, "policy_id")]
 }
 
 # map raw job status to more human readable job status
@@ -173,7 +188,11 @@ job_handles = {
     JT_TRUNK_SYNC: "sync_trunk",
     JT_SHADOW_PORT_SETUP: "setup_shadow_ports",
     JT_SFC_SYNC: "sync_service_function_chain",
-    JT_RESOURCE_RECYCLE: "recycle_resources"
+    JT_RESOURCE_RECYCLE: "recycle_resources",
+    JT_QOS_CREATE: "create_qos_policy",
+    JT_QOS_UPDATE: "update_qos_policy",
+    JT_QOS_DELETE: "delete_qos_policy",
+    JT_SYNC_QOS_RULE: "sync_qos_policy_rules"
 }
 
 # map job type to its primary resource and then we only validate the project_id
@@ -189,7 +208,11 @@ job_primary_resource_map = {
     JT_TRUNK_SYNC: (RT_TRUNK, "trunk_id"),
     JT_SHADOW_PORT_SETUP: (RT_NETWORK, "network_id"),
     JT_SFC_SYNC: (RT_PORT_CHAIN, "portchain_id"),
-    JT_RESOURCE_RECYCLE: (None, "project_id")
+    JT_RESOURCE_RECYCLE: (None, "project_id"),
+    JT_QOS_CREATE: (RT_QOS, "policy_id"),
+    JT_QOS_UPDATE: (RT_QOS, "policy_id"),
+    JT_QOS_DELETE: (RT_QOS, "policy_id"),
+    JT_SYNC_QOS_RULE: (RT_QOS, "policy_id")
 }
 
 # admin API request path

tricircle/common/resource_handle.py

@@ -35,6 +35,9 @@ LIST, CREATE, DELETE, GET, ACTION, UPDATE = 1, 2, 4, 8, 16, 32
 operation_index_map = {'list': LIST, 'create': CREATE, 'delete': DELETE,
                        'get': GET, 'action': ACTION, 'update': UPDATE}
 
+policy_rules = ('bandwidth_limit_rule', 'dscp_marking_rule',
+                'minimum_bandwidth_rule')
+
 LOG = logging.getLogger(__name__)
 
 
@@ -98,7 +101,11 @@ class NeutronResourceHandle(ResourceHandle):
         'port_chain': LIST | CREATE | DELETE | GET | UPDATE,
         'port_pair_group': LIST | CREATE | DELETE | GET | UPDATE,
         'port_pair': LIST | CREATE | DELETE | GET | UPDATE,
-        'flow_classifier': LIST | CREATE | DELETE | GET | UPDATE}
+        'flow_classifier': LIST | CREATE | DELETE | GET | UPDATE,
+        'qos_policy': LIST | CREATE | DELETE | GET | UPDATE,
+        'bandwidth_limit_rule': LIST | CREATE | DELETE | GET | UPDATE,
+        'dscp_marking_rule': LIST | CREATE | DELETE | GET | UPDATE,
+        'minimum_bandwidth_rule': LIST | CREATE | DELETE | GET | UPDATE}
 
     def _get_client(self, cxt):
         token = cxt.auth_token
@@ -113,6 +120,9 @@ class NeutronResourceHandle(ResourceHandle):
     def handle_list(self, cxt, resource, filters):
         try:
             client = self._get_client(cxt)
+            if resource == 'qos_policy':
+                collection = 'qos_policies'
+            else:
                 collection = '%ss' % resource
             search_opts = _transform_filters(filters)
             return [res for res in getattr(
@@ -126,6 +136,10 @@ class NeutronResourceHandle(ResourceHandle):
             client = self._get_client(cxt)
             ret = getattr(client, 'create_%s' % resource)(
                 *args, **kwargs)
+
+            if resource == 'qos_policy':
+                return ret['policy']
+
             if resource in ret:
                 return ret[resource]
             else:
@@ -137,6 +151,9 @@ class NeutronResourceHandle(ResourceHandle):
     def handle_update(self, cxt, resource, *args, **kwargs):
         try:
             client = self._get_client(cxt)
+            if resource == 'qos_policy':
+                return getattr(client, 'update_%s' % resource)(
+                    *args, **kwargs)['policy']
             return getattr(client, 'update_%s' % resource)(
                 *args, **kwargs)[resource]
         except q_exceptions.ConnectionFailed:
@@ -146,6 +163,13 @@ class NeutronResourceHandle(ResourceHandle):
     def handle_get(self, cxt, resource, resource_id):
         try:
             client = self._get_client(cxt)
+            if resource == 'qos_policy':
+                return getattr(client, 'show_%s' % resource)(
+                    resource_id)['policy']
+            if resource in policy_rules:
+                (rule_id, policy_id) = resource_id.split('#')
+                return getattr(client, 'show_%s' % resource)(
+                    rule_id, policy_id)[resource]
             return getattr(client, 'show_%s' % resource)(resource_id)[resource]
         except q_exceptions.ConnectionFailed:
             raise exceptions.EndpointNotAvailable(
@@ -157,6 +181,10 @@ class NeutronResourceHandle(ResourceHandle):
     def handle_delete(self, cxt, resource, resource_id):
         try:
             client = self._get_client(cxt)
+            if resource in policy_rules:
+                (rule_id, policy_id) = resource_id.split('#')
+                return getattr(client, 'delete_%s' % resource)(
+                    rule_id, policy_id)
             return getattr(client, 'delete_%s' % resource)(resource_id)
         except q_exceptions.ConnectionFailed:
             raise exceptions.EndpointNotAvailable(

tricircle/common/xrpcapi.py

@@ -140,3 +140,26 @@ class XJobAPI(object):
             ctxt, project_id,
             constants.job_handles[constants.JT_RESOURCE_RECYCLE],
             constants.JT_RESOURCE_RECYCLE, project_id)
+
+    def create_qos_policy(self, ctxt, project_id, policy_id, pod_id,
+                          res_type, res_id=None):
+        self.invoke_method(
+            ctxt, project_id, constants.job_handles[constants.JT_QOS_CREATE],
+            constants.JT_QOS_CREATE, '%s#%s#%s#%s' % (pod_id, policy_id,
+                                                      res_type, res_id))
+
+    def update_qos_policy(self, ctxt, project_id, policy_id, pod_id):
+        self.invoke_method(
+            ctxt, project_id, constants.job_handles[constants.JT_QOS_UPDATE],
+            constants.JT_QOS_UPDATE, '%s#%s' % (pod_id, policy_id))
+
+    def delete_qos_policy(self, ctxt, project_id, policy_id, pod_id):
+        self.invoke_method(
+            ctxt, project_id, constants.job_handles[constants.JT_QOS_DELETE],
+            constants.JT_QOS_DELETE, '%s#%s' % (pod_id, policy_id))
+
+    def sync_qos_policy_rules(self, ctxt, project_id, policy_id):
+        self.invoke_method(
+            ctxt, project_id,
+            constants.job_handles[constants.JT_SYNC_QOS_RULE],
+            constants.JT_SYNC_QOS_RULE, policy_id)

tricircle/network/central_plugin.py

@@ -28,7 +28,9 @@ from neutron.callbacks import exceptions as callbacks_exc
 from neutron.callbacks import registry
 from neutron.callbacks import resources
 import neutron.common.exceptions as ml2_exceptions
+from neutron.conf.plugins.ml2 import config  # noqa
 from neutron.db import _resource_extend as resource_extend
+from neutron.db import agents_db
 from neutron.db import api as q_db_api
 from neutron.db.availability_zone import router as router_az
 from neutron.db import db_base_plugin_v2
@@ -44,6 +46,8 @@ from neutron.db import l3_hamode_db  # noqa
 from neutron.db import models_v2
 from neutron.db import portbindings_db
 from neutron.extensions import providernet as provider
+from neutron.objects.qos import policy as policy_object
+from neutron.plugins.ml2 import managers as n_managers
 from neutron_lib.api.definitions import availability_zone as az_def
 from neutron_lib.api.definitions import external_net
 from neutron_lib.api.definitions import l3 as l3_apidef
@@ -72,6 +76,7 @@ from tricircle.db import models
 import tricircle.network.exceptions as t_network_exc
 from tricircle.network import helper
 from tricircle.network import managers
+from tricircle.network import qos_driver
 from tricircle.network import security_groups
 
 
@@ -80,6 +85,11 @@ tricircle_opts = [
                 default=['vxlan,local'],
                 help=_('List of network type driver entry points to be loaded '
                        'from the tricircle.network.type_drivers namespace.')),
+    cfg.ListOpt('extension_drivers',
+                default=[],
+                help=_('List of network extension driver entry points to be '
+                       'loaded from the neutron.ml2.extension_drivers '
+                       'namespace.')),
     cfg.ListOpt('tenant_network_types',
                 default=['vxlan,local'],
                 help=_('Ordered list of network_types to allocate as tenant '
@@ -128,6 +138,7 @@ NON_VM_PORT_TYPES = [constants.DEVICE_OWNER_ROUTER_INTF,
 
 
 class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
+                      agents_db.AgentDbMixin,
                       security_groups.TricircleSecurityGroupMixin,
                       external_net_db.External_net_db_mixin,
                       portbindings_db.PortBindingMixin,
@@ -165,12 +176,15 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
     def __init__(self):
         super(TricirclePlugin, self).__init__()
         LOG.info("Starting Tricircle Neutron Plugin")
-        self.clients = {}
+        self.clients = {'top': t_client.Client()}
         self.xjob_handler = xrpcapi.XJobAPI()
         self._setup_rpc()
         self.type_manager = managers.TricircleTypeManager()
+        self.extension_manager = n_managers.ExtensionManager()
+        self.extension_manager.initialize()
         self.type_manager.initialize()
         self.helper = helper.NetworkHelper(self)
+        qos_driver.register()
 
     def _setup_rpc(self):
         self.endpoints = []
@@ -303,6 +317,8 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             net_db = self.create_network_db(context, network)
             res = self._make_network_dict(net_db, process_extensions=False,
                                           context=context)
+            self.extension_manager.process_create_network(context, net_data,
+                                                          res)
             self._process_l3_create(context, res, net_data)
             net_data['id'] = res['id']
             self.type_manager.create_network_segments(context, net_data,
@@ -392,19 +408,55 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         self._raise_if_updates_external_attribute(net_data)
 
         with context.session.begin():
-            net = super(TricirclePlugin, self).update_network(context,
-                                                              network_id,
-                                                              network)
+            original_network = super(TricirclePlugin, self).get_network(
+                context, network_id)
+            policy = policy_object.QosPolicy.get_network_policy(
+                context, network_id)
+            if policy:
+                original_network['qos_policy_id'] = policy['id']
+            else:
+                original_network['qos_policy_id'] = None
+
+            updated_network = super(
+                TricirclePlugin, self).update_network(
+                context, network_id, network)
+
+            self.extension_manager.process_update_network(
+                context, net_data, updated_network)
+
+            self.type_manager.extend_network_dict_provider(context,
+                                                           updated_network)
+
+            updated_network = self.get_network(context, network_id)
+
+            if net_data.get('qos_policy_id', None):
+                updated_network['qos_policy_id'] = net_data['qos_policy_id']
+
+            if not updated_network.get('qos_policy_id', None):
+                updated_network['qos_policy_id'] = None
+
+            need_network_update_notify = (
+                'qos_policy_id' in net_data and
+                original_network['qos_policy_id'] !=
+                updated_network['qos_policy_id'])
+
             t_ctx = t_context.get_context_from_neutron_context(context)
             mappings = db_api.get_bottom_mappings_by_top_id(
                 t_ctx, network_id, t_constants.RT_NETWORK)
             if mappings:
                 self.xjob_handler.update_network(
-                    t_ctx, net['tenant_id'], network_id,
+                    t_ctx, updated_network['tenant_id'], network_id,
                     t_constants.POD_NOT_SPECIFIED)
 
-            self.type_manager.extend_network_dict_provider(context, net)
-            return net
+            if need_network_update_notify and \
+                    updated_network['qos_policy_id'] and mappings:
+                t_policy_id = updated_network['qos_policy_id']
+                self.xjob_handler.create_qos_policy(
+                    t_ctx, updated_network['tenant_id'], t_policy_id,
+                    t_constants.POD_NOT_SPECIFIED, t_constants.RT_NETWORK,
+                    updated_network['id'])
+
+            return updated_network
 
     def _convert_az2region_for_nets(self, context, nets):
         for net in nets:
@@ -419,6 +471,11 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
     def _convert_az2region(self, t_ctx, az_hints):
         return self.helper.convert_az2region(t_ctx, az_hints)
 
+    def _get_network_qos_info(self, context, net_id):
+        policy = policy_object.QosPolicy.get_network_policy(
+            context.elevated(), net_id)
+        return policy['id'] if policy else None
+
     def get_network(self, context, network_id, fields=None):
         net = super(TricirclePlugin, self).get_network(context, network_id,
                                                        fields)
@@ -427,6 +484,9 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
 
         self._convert_az2region_for_net(context, net)
 
+        net['qos_policy_id'] = \
+            self._get_network_qos_info(context.elevated(), net['id'])
+
         return net
 
     def get_networks(self, context, filters=None, fields=None,
@@ -438,6 +498,11 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             self.type_manager.extend_networks_dict_provider(context, nets)
 
         self._convert_az2region_for_nets(context, nets)
+
+        for net in nets:
+            net['qos_policy_id'] = \
+                self._get_network_qos_info(context.elevated(), net['id'])
+
         return nets
 
     def create_subnet(self, context, subnet):
@@ -561,6 +626,7 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         self._ensure_default_security_group_on_port(context, port)
         sgids = self._get_security_groups_on_port(context, port)
         result = self._make_port_dict(db_port)
+        self.extension_manager.process_create_port(context, port_body, result)
         self._process_port_create_security_group(context, result, sgids)
         return result
 
@@ -671,7 +737,7 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
     def update_port(self, context, port_id, port):
         t_ctx = t_context.get_context_from_neutron_context(context)
         top_port = super(TricirclePlugin, self).get_port(context, port_id)
-
+        updated_port = None
         # be careful that l3_db will call update_port to update device_id of
         # router interface, we cannot directly update bottom port in this case,
         # otherwise we will fail when attaching bottom port to bottom router
@@ -679,15 +745,17 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         if t_constants.PROFILE_REGION in port['port'].get(
                 'binding:profile', {}):
             # this update request comes from local Neutron
-            res = super(TricirclePlugin, self).update_port(context, port_id,
+            updated_port = super(TricirclePlugin, self).update_port(context,
+                                                                    port_id,
                                                                     port)
+
             profile_dict = port['port']['binding:profile']
             region_name = profile_dict[t_constants.PROFILE_REGION]
             device_name = profile_dict[t_constants.PROFILE_DEVICE]
             t_ctx = t_context.get_context_from_neutron_context(context)
             pod = db_api.get_pod_by_name(t_ctx, region_name)
 
-            net = self.get_network(context, res['network_id'])
+            net = self.get_network(context, updated_port['network_id'])
             is_vxlan_network = (
                 net[provider_net.NETWORK_TYPE] == t_constants.NT_VxLAN)
             if is_vxlan_network:
@@ -700,20 +768,41 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
                 # gateway port, but we only need to create resource routing
                 # entries, trigger xjob and configure security group rules for
                 # instance port
-                self._create_mapping_for_vm_port(t_ctx, res, pod)
+                self._create_mapping_for_vm_port(t_ctx, updated_port, pod)
                 self._process_trunk_port(context, t_ctx,
-                                         res, pod, profile_dict)
+                                         updated_port, pod, profile_dict)
                 # only trigger setup_bottom_router job
-                self._trigger_router_xjob_for_vm_port(context, res, pod)
+                self._trigger_router_xjob_for_vm_port(context, updated_port,
+                                                      pod)
                 self.xjob_handler.configure_security_group_rules(
-                    t_ctx, res['tenant_id'])
+                    t_ctx, updated_port['tenant_id'])
 
             if is_vxlan_network and (
                     cfg.CONF.client.cross_pod_vxlan_mode in (
                         t_constants.NM_P2P, t_constants.NM_L2GW)):
-                self.xjob_handler.setup_shadow_ports(t_ctx, res['tenant_id'],
-                                                     pod['pod_id'],
-                                                     res['network_id'])
+                self.xjob_handler.setup_shadow_ports(
+                    t_ctx, updated_port['tenant_id'], pod['pod_id'],
+                    updated_port['network_id'])
+
+            network_binding_policy = \
+                policy_object.QosPolicy.get_network_policy(
+                    context, updated_port['network_id'])
+
+            port_binding_policy = policy_object.QosPolicy.get_port_policy(
+                context, port_id)
+
+            if network_binding_policy:
+                t_policy_id = network_binding_policy['id']
+                self.xjob_handler.create_qos_policy(
+                    t_ctx, t_ctx.project_id, t_policy_id, pod['pod_id'],
+                    t_constants.RT_NETWORK, updated_port['network_id'])
+
+            if port_binding_policy:
+                t_policy_id = port_binding_policy['id']
+                self.xjob_handler.create_qos_policy(
+                    t_ctx, t_ctx.project_id, t_policy_id, pod['pod_id'],
+                    t_constants.RT_PORT, port_id)
+
         # for vm port or port with empty device_owner, update top port and
         # bottom port
         elif top_port.get('device_owner') not in NON_VM_PORT_TYPES:
@@ -722,6 +811,9 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             request_body = port[attributes.PORT]
             if mappings:
                 with context.session.begin():
+                    original_qos_policy_id = \
+                        self._get_port_qos_info(context, port_id)
+
                     b_pod, b_port_id = mappings[0]
                     b_region_name = b_pod['region_name']
                     b_client = self._get_client(region_name=b_region_name)
@@ -733,14 +825,24 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
                         self._handle_bottom_security_group(
                             t_ctx, request_body['security_groups'], b_pod)
 
-                    res = super(TricirclePlugin, self).update_port(
+                    updated_port = super(TricirclePlugin, self).update_port(
                         context, port_id, port)
+                    self.extension_manager.process_update_port(
+                        context, request_body, updated_port)
+                    updated_port = \
+                        super(TricirclePlugin, self).get_port(context, port_id)
                     # name is not allowed to be updated, because it is used by
                     # lock_handle to retrieve bottom/local resources that have
                     # been created but not registered in the resource routing
                     # table
                     request_body.pop('name', None)
 
+                    request_body_policy_id = \
+                        request_body.get('qos_policy_id', None)
+                    if request_body_policy_id:
+                            request_body.pop('qos_policy_id')
+
+                    if request_body:
                         try:
                             b_client.update_ports(t_ctx, b_port_id, port)
                         except q_cli_exceptions.NotFound:
@@ -751,18 +853,30 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
 
                     if request_body.get('security_groups', None):
                         self.xjob_handler.configure_security_group_rules(
-                            t_ctx, res['tenant_id'])
+                            t_ctx, updated_port['tenant_id'])
+
+                    updated_port['qos_policy_id'] = request_body_policy_id
+                    if request_body_policy_id and \
+                        original_qos_policy_id != \
+                            request_body_policy_id:
+                        t_policy_id = updated_port['qos_policy_id']
+                        self.xjob_handler.create_qos_policy(
+                            t_ctx, t_ctx.project_id,
+                            t_policy_id, b_pod['pod_id'],
+                            t_constants.RT_PORT, b_port_id)
             else:
                 self._filter_unsupported_attrs(request_body)
-                res = super(TricirclePlugin, self).update_port(
+                updated_port = super(TricirclePlugin, self).update_port(
                     context, port_id, port)
+                self.extension_manager.process_update_port(
+                    context, request_body, updated_port)
         else:
             # for router interface, router gw, dhcp port, not directly
             # update bottom port
-            res = super(TricirclePlugin, self).update_port(
+            updated_port = super(TricirclePlugin, self).update_port(
                 context, port_id, port)
         self._log_update_port_sensitive_attrs(port_id, port)
-        return res
+        return updated_port
 
     def _pre_delete_port(self, context, port_id, port_check):
         """Do some preliminary operations before deleting the port."""
@@ -816,6 +930,10 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
                                                 'value': port_id}])
         super(TricirclePlugin, self).delete_port(context, port_id)
 
+    def _get_port_qos_info(self, context, port_id):
+        policy = policy_object.QosPolicy.get_port_policy(context, port_id)
+        return policy['id'] if policy else None
+
     def get_port(self, context, port_id, fields=None):
         t_ctx = t_context.get_context_from_neutron_context(context)
         mappings = db_api.get_bottom_mappings_by_top_id(
@@ -830,12 +948,11 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             if fields:
                 port = dict(
                     [(k, v) for k, v in six.iteritems(port) if k in fields])
-            if 'network_id' not in port and 'fixed_ips' not in port:
-                return port
-
+            if 'network_id' in port or 'fixed_ips' in port:
                 bottom_top_map = {}
                 with t_ctx.session.begin():
-                for resource in (t_constants.RT_SUBNET, t_constants.RT_NETWORK,
+                    for resource in (t_constants.RT_SUBNET,
+                                     t_constants.RT_NETWORK,
                                      t_constants.RT_ROUTER):
                         route_filters = [{'key': 'resource_type',
                                           'comparator': 'eq',
@@ -847,11 +964,14 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
                                 bottom_top_map[
                                     route['bottom_id']] = route['top_id']
                 self._map_port_from_bottom_to_top(port, bottom_top_map)
-            return port
         else:
-            return super(TricirclePlugin, self).get_port(context,
+            port = super(TricirclePlugin, self).get_port(context,
                                                          port_id, fields)
 
+        port['qos_policy_id'] = \
+            self._get_port_qos_info(context, port_id)
+        return port
+
     @staticmethod
     def _apply_ports_filters(query, model, filters):
         if not filters:
@@ -1051,8 +1171,13 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         if not filters or 'id' not in filters:
             # if filter is empty or "id" is not in the filter, no special
             # handle is required
-            return self._get_ports(context, filters, fields, sorts, limit,
+            ports = self._get_ports(context, filters, fields, sorts, limit,
                                     marker, page_reverse)
+            for port in ports:
+                port['qos_policy_id'] = \
+                    self._get_port_qos_info(context, port['id'])
+
+            return ports
         if len(filters) == 1:
             # only "id" is in the filter, we use get_port to get all the ports
             ports = []
@@ -1068,9 +1193,14 @@ class TricirclePlugin(db_base_plugin_v2.NeutronDbPluginV2,
             id_filters = filters.pop('id')
             ports = self._get_ports(context, filters, None, sorts, limit,
                                     marker, page_reverse)
-            return [super(TricirclePlugin,
+            ports = [super(TricirclePlugin,
                            self)._fields(
                 p, fields) for p in ports if p['id'] in id_filters]
+            for port in ports:
+                port['qos_policy_id'] = \
+                    self._get_port_qos_info(context, port['id'])
+
+            return ports
 
     def _get_ports(self, context, filters=None, fields=None, sorts=None,
                    limit=None, marker=None, page_reverse=False):

tricircle/network/central_qos_plugin.py

@@ -0,0 +1,83 @@
+# Copyright 2017 Hunan University.
+# 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.objects import ports as ports_object
+from neutron.services.qos import qos_plugin
+from neutron_lib.api.definitions import portbindings
+from oslo_log import log
+
+import tricircle.common.client as t_client
+import tricircle.common.constants as t_constants
+import tricircle.common.context as t_context
+import tricircle.db.api as db_api
+
+LOG = log.getLogger(__name__)
+
+
+class TricircleQosPlugin(qos_plugin.QoSPlugin):
+
+    def __init__(self):
+        super(TricircleQosPlugin, self).__init__()
+        self.clients = {'top': t_client.Client()}
+
+    def _get_client(self, region_name):
+        if region_name not in self.clients:
+            self.clients[region_name] = t_client.Client(region_name)
+        return self.clients[region_name]
+
+    def _get_ports_with_policy(self, context, policy):
+        networks_ids = policy.get_bound_networks()
+
+        ports_with_net_policy = ports_object.Port.get_objects(
+            context, network_id=networks_ids)
+
+        # Filter only these ports which don't have overwritten policy
+        ports_with_net_policy = [
+            port for port in ports_with_net_policy if
+            port.qos_policy_id is None
+        ]
+
+        ports_ids = policy.get_bound_ports()
+        ports_with_policy = ports_object.Port.get_objects(
+            context, id=ports_ids)
+        t_ports = list(set(ports_with_policy + ports_with_net_policy))
+
+        t_ctx = t_context.get_context_from_neutron_context(context)
+        for t_port in t_ports:
+            mappings = db_api.get_bottom_mappings_by_top_id(
+                t_ctx, t_port.id, t_constants.RT_PORT)
+            if mappings:
+                b_pod, b_port_id = mappings[0]
+                b_region_name = b_pod['region_name']
+                b_client = self._get_client(region_name=b_region_name)
+                b_port = b_client.get_ports(t_ctx, b_port_id)
+                new_binding = ports_object.PortBinding(
+                    port_id=t_port.id,
+                    vif_type=b_port.get('binding:vif_type',
+                                        portbindings.VIF_TYPE_UNBOUND),
+                    vnic_type=b_port.get('binding:vnic_type',
+                                         portbindings.VNIC_NORMAL)
+                )
+                t_port.binding = new_binding
+            else:
+                new_binding = ports_object.PortBinding(
+                    port_id=t_port.id,
+                    vif_type=portbindings.VIF_TYPE_UNBOUND,
+                    vnic_type=portbindings.VNIC_NORMAL
+                )
+                t_port.binding = new_binding
+
+        return t_ports

tricircle/network/local_plugin.py

@@ -245,6 +245,9 @@ class TricirclePlugin(plugin.Ml2Plugin):
                                               net_body['name'])
             if net_id:
                 net_body['id'] = net_id
+
+        net_body.pop('qos_policy_id', None)
+
         b_network = self.core_plugin.create_network(context,
                                                     {'network': net_body})
         return b_network
@@ -349,6 +352,8 @@ class TricirclePlugin(plugin.Ml2Plugin):
                     continue
 
                 self._adapt_network_body(network)
+
+                network.pop('qos_policy_id', None)
                 b_network = self.core_plugin.create_network(
                     context, {'network': network})
                 subnet_ids = self._ensure_subnet(context, network)
@@ -391,6 +396,7 @@ class TricirclePlugin(plugin.Ml2Plugin):
         if not t_network:
             raise q_exceptions.NetworkNotFound(net_id=_id)
         self._adapt_network_body(t_network)
+        t_network.pop('qos_policy_id', None)
         b_network = self.core_plugin.create_network(context,
                                                     {'network': t_network})
         return t_network, b_network
@@ -594,6 +600,8 @@ class TricirclePlugin(plugin.Ml2Plugin):
 
         self._handle_security_group(t_ctx, context, t_port)
         self._create_shadow_agent(context, port_body)
+
+        t_port.pop('qos_policy_id', None)
         b_port = self.core_plugin.create_port(context, {'port': t_port})
         return b_port
 
@@ -779,6 +787,7 @@ class TricirclePlugin(plugin.Ml2Plugin):
             self._ensure_network_subnet(context, t_port)
             self._adapt_port_body_for_call(t_port)
             self._handle_security_group(t_ctx, context, t_port)
+            t_port.pop('qos_policy_id', None)
             b_port = self.core_plugin.create_port(context, {'port': t_port})
 
         self._ensure_trunk(context, t_ctx, _id)
@@ -822,6 +831,7 @@ class TricirclePlugin(plugin.Ml2Plugin):
             self._ensure_network_subnet(context, port)
             self._adapt_port_body_for_call(port)
             self._handle_security_group(t_ctx, context, port)
+            port.pop('qos_policy_id', None)
             b_port = self.core_plugin.create_port(context,
                                                   {'port': port})
             b_ports.append(self._fields(b_port, fields))

tricircle/network/qos_driver.py

@@ -0,0 +1,144 @@
+# Copyright 2017 Hunan University Co., 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 neutron_lib.api.definitions import portbindings
+from neutron_lib import constants
+from neutron_lib.db import constants as db_constants
+from neutron_lib.services.qos import base
+from neutron_lib.services.qos import constants as qos_consts
+
+from oslo_log import log as logging
+
+from tricircle.common import constants as t_constants
+from tricircle.common import context
+from tricircle.common import xrpcapi
+from tricircle.db import api as db_api
+
+LOG = logging.getLogger(__name__)
+
+DRIVER = None
+
+SUPPORTED_RULES = {
+    qos_consts.RULE_TYPE_BANDWIDTH_LIMIT: {
+        qos_consts.MAX_KBPS: {
+            'type:range': [0, db_constants.DB_INTEGER_MAX_VALUE]},
+        qos_consts.MAX_BURST: {
+            'type:range': [0, db_constants.DB_INTEGER_MAX_VALUE]},
+        qos_consts.DIRECTION: {
+            'type:values': constants.VALID_DIRECTIONS}
+    },
+    qos_consts.RULE_TYPE_DSCP_MARKING: {
+        qos_consts.DSCP_MARK: {'type:values': constants.VALID_DSCP_MARKS}
+    },
+    qos_consts.RULE_TYPE_MINIMUM_BANDWIDTH: {
+        qos_consts.MIN_KBPS: {
+            'type:range': [0, db_constants.DB_INTEGER_MAX_VALUE]},
+        qos_consts.DIRECTION: {'type:values': [constants.EGRESS_DIRECTION]}
+    }
+}
+
+VIF_TYPES = [portbindings.VIF_TYPE_OVS,
+             portbindings.VIF_TYPE_VHOST_USER,
+             portbindings.VIF_TYPE_UNBOUND]
+
+
+class TricircleQoSDriver(base.DriverBase):
+    def __init__(self, name, vif_types, vnic_types,
+                 supported_rules,
+                 requires_rpc_notifications):
+        super(TricircleQoSDriver, self).__init__(name, vif_types, vnic_types,
+                                                 supported_rules,
+                                                 requires_rpc_notifications)
+        self.xjob_handler = xrpcapi.XJobAPI()
+
+    @staticmethod
+    def create():
+        return TricircleQoSDriver(
+            name='tricircle',
+            vif_types=VIF_TYPES,
+            vnic_types=portbindings.VNIC_TYPES,
+            supported_rules=SUPPORTED_RULES,
+            requires_rpc_notifications=False)
+
+    def create_policy(self, q_context, policy):
+        """Create policy invocation.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being created, which will have no
+                      rules.
+        """
+        pass
+
+    def create_policy_precommit(self, q_context, policy):
+        """Create policy precommit.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being created, which will have no
+                      rules.
+        """
+        pass
+
+    def update_policy(self, q_context, policy):
+        """Update policy invocation.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being updated.
+        """
+        pass
+
+    def update_policy_precommit(self, q_context, policy):
+        """Update policy precommit.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being updated.
+        """
+        t_context = context.get_context_from_neutron_context(q_context)
+        policy_id = policy['id']
+        mappings = db_api.get_bottom_mappings_by_top_id(
+            t_context, policy_id, t_constants.RT_QOS)
+
+        if mappings:
+            self.xjob_handler.update_qos_policy(
+                t_context, t_context.project_id, policy_id,
+                t_constants.POD_NOT_SPECIFIED)
+            self.xjob_handler.sync_qos_policy_rules(
+                t_context, t_context.project_id, policy_id)
+
+    def delete_policy(self, q_context, policy):
+        """Delete policy invocation.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being deleted
+        """
+
+    def delete_policy_precommit(self, q_context, policy):
+        """Delete policy precommit.
+
+        :param q_context: current running context information
+        :param policy: a QoSPolicy object being deleted
+        """
+        t_context = context.get_context_from_neutron_context(q_context)
+        policy_id = policy['id']
+        self.xjob_handler.delete_qos_policy(
+            t_context, t_context.project_id, policy_id,
+            t_constants.POD_NOT_SPECIFIED)
+
+
+def register():
+    """Register the driver."""
+    global DRIVER
+    if not DRIVER:
+        DRIVER = TricircleQoSDriver.create()
+    LOG.debug('Tricircle QoS driver registered')

tricircle/tempestplugin/gate_hook.sh

@@ -48,12 +48,15 @@ function _setup_tricircle_multinode {
     export DEVSTACK_LOCAL_CONFIG+=$'\n'"TRICIRCLE_START_SERVICES=True"
     export DEVSTACK_LOCAL_CONFIG+=$'\n'"TRICIRCLE_ENABLE_TRUNK=True"
     export DEVSTACK_LOCAL_CONFIG+=$'\n'"TRICIRCLE_ENABLE_SFC=True"
+    export DEVSTACK_LOCAL_CONFIG+=$'\n'"TRICIRCLE_ENABLE_QOS=True"
     export DEVSTACK_LOCAL_CONFIG+=$'\n'"REGION_NAME=RegionOne"
     export DEVSTACK_LOCAL_CONFIG+=$'\n'"HOST_IP=$PRIMARY_NODE_IP"
 
     ML2_CONFIG=$'\n'"ML2_L3_PLUGIN=tricircle.network.local_l3_plugin.TricircleL3Plugin"
+    ML2_CONFIG+=$'\n'"ML2_L3_PLUGIN+=,neutron.services.qos.qos_plugin.QoSPlugin"
     ML2_CONFIG+=$'\n'"[[post-config|/"'$Q_PLUGIN_CONF_FILE]]'
     ML2_CONFIG+=$'\n'"[ml2]"
+    ML2_CONFIG+=$'\n'"extension_drivers = port_security,qos"
     ML2_CONFIG+=$'\n'"mechanism_drivers = openvswitch,linuxbridge,l2population"
     ML2_CONFIG+=$'\n'"[agent]"
     ML2_CONFIG+=$'\n'"extensions=sfc"

tricircle/tempestplugin/qos_policy_rule_test.yaml

@@ -0,0 +1,861 @@
+- task_set_id: preparation
+  tasks:
+    - task_id: policy1
+      type: qos_policy
+      region: central
+      params:
+        name: policy1
+    - task_id: bandwidth_limit_rule1
+      region: central
+      type: qos_bandwidth_limit_rule
+      depend: [policy1]
+      params:
+        max_kbps: 3000
+        max_burst_kbps: 300
+        qos_policy: policy1@id
+    - task_id: policy2
+      type: qos_policy
+      region: central
+      params:
+        name: policy2
+    - task_id: bandwidth_limit_rule2
+      region: central
+      type: qos_bandwidth_limit_rule
+      depend: [policy2]
+      params:
+        max_kbps: 3000
+        max_burst_kbps: 300
+        qos_policy: policy2@id
+    - task_id: policy3
+      type: qos_policy
+      region: central
+      params:
+        name: policy3
+    - task_id: policy4
+      type: qos_policy
+      region: central
+      params:
+        name: policy4
+    - task_id: policy5
+      type: qos_policy
+      region: central
+      params:
+        name: policy5
+    - task_id: bandwidth_limit_rule5
+      region: central
+      type: qos_bandwidth_limit_rule
+      depend: [policy5]
+      params:
+        max_kbps: 3000
+        max_burst_kbps: 300
+        qos_policy: policy5@id
+    - task_id: dscp_marking_rule1
+      region: central
+      type: qos_dscp_marking_rule
+      depend: [policy1]
+      params:
+        dscp_mark: 30
+        qos_policy: policy1@id
+    - task_id: net1
+      region: central
+      type: network
+      params:
+        name: net1
+    - task_id: subnet1
+      region: central
+      type: subnet
+      depend: [net1]
+      params:
+        name: subnet1
+        ip_version: 4
+        cidr: 10.0.1.0/24
+        network_id: net1@id
+    - task_id: port1
+      region: central
+      type: port
+      depend:
+        - net1
+        - subnet1
+      params:
+        name: port1
+        network_id: net1@id
+    - task_id: net2
+      region: central
+      type: network
+      params:
+        name: net2
+    - task_id: subnet2
+      region: central
+      type: subnet
+      depend: [net2]
+      params:
+        name: subnet2
+        ip_version: 4
+        cidr: 10.0.2.0/24
+        network_id: net2@id
+    - task_id: port2
+      region: central
+      type: port
+      depend:
+        - net2
+        - subnet2
+      params:
+        name: port2
+        network_id: net2@id
+    - task_id: net3
+      region: central
+      type: network
+      params:
+        name: net3
+    - task_id: subnet3
+      region: central
+      type: subnet
+      depend: [net3]
+      params:
+        name: subnet3
+        ip_version: 4
+        cidr: 10.0.3.0/24
+        network_id: net3@id
+    - task_id: port3
+      region: central
+      type: port
+      depend:
+        - net3
+        - subnet3
+      params:
+        name: port3
+        network_id: net3@id
+    - task_id: net4
+      region: central
+      type: network
+      params:
+        name: net4
+    - task_id: subnet4
+      region: central
+      type: subnet
+      depend: [net4]
+      params:
+        name: subnet4
+        ip_version: 4
+        cidr: 10.0.4.0/24
+        network_id: net4@id
+    - task_id: port4
+      region: central
+      type: port
+      depend:
+        - net4
+        - subnet4
+      params:
+        name: port4
+        network_id: net4@id
+    - task_id: net5
+      region: central
+      type: network
+      params:
+        name: net5
+    - task_id: image1
+      region: region1
+      type: image
+      query:
+        get_one: true
+- task_set_id: check_qos_create
+  depend: [preparation]
+  tasks:
+    - task_id: check_policy1_central
+      region: central
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy1
+    - task_id: check_bandwidth_limit_rule1
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: any
+        condition:
+          - id: preparation@bandwidth_limit_rule1@id
+    - task_id: check_dscp_marking_rule1
+      region: central
+      type: qos_dscp_marking_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: any
+        condition:
+          - id: preparation@dscp_marking_rule1@id
+    - task_id: check_policy1_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: all
+        condition:
+          - name: invalid-name
+- task_set_id: policy_update_only_central
+  depend: [preparation]
+  tasks:
+    - task_id: policy1_update_only_central
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy1@id
+        method: update
+      params:
+        name: policy1_update_only_central
+    - task_id: bandwidth_limit_rule1_update_only_central
+      region: central
+      type: qos_bandwidth_limit_rule
+      action:
+        target: preparation@bandwidth_limit_rule1@id
+        method: update
+      params:
+        qos_policy: preparation@policy1@id
+        max_kbps: 4000
+    - task_id: dscp_marking_rule1_update_only_central
+      region: central
+      type: qos_dscp_marking_rule
+      action:
+        target: preparation@dscp_marking_rule1@id
+        method: update
+      params:
+        qos_policy: preparation@policy1@id
+        dscp_mark: 40
+- task_set_id: check_qos_update_only_central
+  depend: [preparation]
+  tasks:
+    - task_id: check_policy1_update_only_central
+      region: central
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy1_update_only_central
+    - task_id: check_limit_rule1_update_only_central
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: any
+        condition:
+          - id: preparation@bandwidth_limit_rule1@id
+            max_kbps: 4000
+    - task_id: check_dscp_rule1_update_only_central
+      region: central
+      type: qos_dscp_marking_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: any
+        condition:
+          - id: preparation@dscp_marking_rule1@id
+            dscp_mark: 40
+- task_set_id: central_bound_policy
+  depend: [preparation]
+  tasks:
+    - task_id: net1_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net1@id
+        method: update
+      params:
+        qos_policy_id: preparation@policy1@id
+    - task_id: net5_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net5@id
+        method: update
+      params:
+        qos_policy_id: preparation@policy5@id
+    - task_id: port3_policy
+      region: central
+      type: port
+      action:
+        target: preparation@port3@id
+        method: update
+      params:
+        qos_policy_id: preparation@policy3@id
+- task_set_id: create_vm
+  depend: [preparation]
+  tasks:
+    - task_id: vm1
+      region: region1
+      type: server
+      params:
+        flavor_id: 1
+        image_id: preparation@image1@id
+        name: vm1
+        networks:
+          - uuid: preparation@net1@id
+            port: preparation@port1@id
+    - task_id: vm2
+      region: region1
+      type: server
+      params:
+        flavor_id: 1
+        image_id: preparation@image1@id
+        name: vm2
+        networks:
+          - uuid: preparation@net2@id
+    - task_id: vm3
+      region: region1
+      type: server
+      params:
+        flavor_id: 1
+        image_id: preparation@image1@id
+        name: vm3
+        networks:
+          - uuid: preparation@net3@id
+            port: preparation@port3@id
+    - task_id: vm4
+      region: region1
+      type: server
+      params:
+        flavor_id: 1
+        image_id: preparation@image1@id
+        name: vm4
+        networks:
+          - uuid: preparation@net4@id
+            port: preparation@port4@id
+- task_set_id: check_vm
+  depend: [preparation]
+  tasks:
+    - task_id: check_vm1
+      region: region1
+      type: server
+      validate:
+        predicate: any
+        retries: 10
+        condition:
+          - status: ACTIVE
+            name: vm1
+    - task_id: check_vm2
+      region: region1
+      type: server
+      validate:
+        predicate: any
+        retries: 10
+        condition:
+          - status: ACTIVE
+            name: vm2
+    - task_id: check_vm3
+      region: region1
+      type: server
+      validate:
+        predicate: any
+        retries: 10
+        condition:
+          - status: ACTIVE
+            name: vm3
+    - task_id: check_vm4
+      region: region1
+      type: server
+      validate:
+        predicate: any
+        retries: 10
+        condition:
+          - status: ACTIVE
+            name: vm4
+- task_set_id: wait_for_vm
+  tasks:
+    - task_id: check_job_vm
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS
+- task_set_id: local_bound_policy
+  depend: [preparation]
+  tasks:
+    - task_id: net2_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net2@id
+        method: update
+      params:
+        qos_policy_id: preparation@policy2@id
+    - task_id: port4_policy
+      region: central
+      type: port
+      action:
+        target: preparation@port4@id
+        method: update
+      params:
+        qos_policy_id: preparation@policy4@id
+- task_set_id: wait_for_bound
+  tasks:
+    - task_id: check_job_bound
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS
+- task_set_id: check_bound_policy
+  depend: [preparation]
+  tasks:
+    - task_id: check_net1_policy_central
+      region: central
+      type: network
+      validate:
+        predicate: any
+        condition:
+          - qos_policy_id: preparation@policy1@id
+    - task_id: check_net2_policy_central
+      region: central
+      type: network
+      validate:
+        predicate: any
+        condition:
+          - qos_policy_id: preparation@policy2@id
+    - task_id: check_net5_policy_central
+      region: central
+      type: network
+      validate:
+        predicate: any
+        condition:
+          - qos_policy_id: preparation@policy5@id
+    - task_id: check_port3_policy_central
+      region: central
+      type: port
+      validate:
+        predicate: any
+        condition:
+          - qos_policy_id: preparation@policy3@id
+    - task_id: check_port4_policy_central
+      region: central
+      type: port
+      validate:
+        predicate: any
+        condition:
+          - qos_policy_id: preparation@policy4@id
+    - task_id: check_policy1_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy1_update_only_central
+    - task_id: check_policy2_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy2
+    - task_id: check_policy3_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy3
+    - task_id: check_policy4_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy4
+- task_set_id: policy_update_with_local
+  depend: [preparation]
+  tasks:
+    - task_id: policy4_update_with_local
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy4@id
+        method: update
+      params:
+        name: policy4_update_with_local
+    - task_id: bandwidth_limit_rule2_update_with_local
+      region: central
+      type: qos_bandwidth_limit_rule
+      action:
+        target: preparation@bandwidth_limit_rule2@id
+        method: update
+      params:
+        qos_policy: preparation@policy2@id
+        max_kbps: 5000
+- task_set_id: wait_for_job_update
+  tasks:
+    - task_id: check_job_update
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS
+- task_set_id: check_qos_update_with_local
+  depend: [preparation]
+  tasks:
+    - task_id: check_policy4_update_with_local
+      region: central
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy4_update_with_local
+    - task_id: check_policy4_update_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: any
+        condition:
+          - name: policy4_update_with_local
+    - task_id: check_limit_rule2_update_with_local
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy2@id
+      validate:
+        predicate: any
+        condition:
+          - id: preparation@bandwidth_limit_rule2@id
+            max_kbps: 5000
+- task_set_id: unbound_policy
+  depend: [preparation]
+  tasks:
+    - task_id: net1_no_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net1@id
+        method: update
+      params:
+        qos_policy_id:
+    - task_id: net2_no_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net2@id
+        method: update
+      params:
+        qos_policy_id:
+    - task_id: port3_no_policy
+      region: central
+      type: port
+      action:
+        target: preparation@port3@id
+        method: update
+      params:
+        qos_policy_id:
+    - task_id: port4_no_policy
+      region: central
+      type: port
+      action:
+        target: preparation@port4@id
+        method: update
+      params:
+        qos_policy_id:
+    - task_id: net5_no_policy
+      region: central
+      type: network
+      action:
+        target: preparation@net5@id
+        method: update
+      params:
+        qos_policy_id:
+- task_set_id: wait_for_qos_unbound
+  tasks:
+    - task_id: check_job_qos_unbound
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS
+- task_set_id: qos-rule-delete
+  depend: [preparation]
+  tasks:
+    - task_id: bandwidth_limit_rule1_delete
+      region: central
+      type: qos_bandwidth_limit_rule
+      action:
+        target: preparation@bandwidth_limit_rule1@id
+        method: delete
+      params:
+        qos_policy: preparation@policy1@id
+    - task_id: bandwidth_limit_rule2_delete
+      region: central
+      type: qos_bandwidth_limit_rule
+      action:
+        target: preparation@bandwidth_limit_rule2@id
+        method: delete
+      params:
+        qos_policy: preparation@policy2@id
+    - task_id: dscp_marking_rule1_delete
+      region: central
+      type: qos_dscp_marking_rule
+      action:
+        target: preparation@dscp_marking_rule1@id
+        method: delete
+      params:
+        qos_policy: preparation@policy1@id
+    - task_id: bandwidth_limit_rule5_delete
+      region: central
+      type: qos_bandwidth_limit_rule
+      action:
+        target: preparation@bandwidth_limit_rule5@id
+        method: delete
+      params:
+        qos_policy: preparation@policy5@id
+- task_set_id: wait_for_rule_delete
+  depend: [preparation]
+  tasks:
+    - task_id: check_job_rule_delete
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS
+    - task_id: check_for_bandwidth_limit1_delete_central
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            id: invalid-id
+    - task_id: check_for_bandwidth_limit2_delete_central
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy2@id
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            id: invalid-id
+    - task_id: check_for_bandwidth_limit5_delete_central
+      region: central
+      type: qos_bandwidth_limit_rule
+      params:
+        qos_policy: preparation@policy5@id
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            id: invalid-id
+    - task_id: check_for_dscp_marking1_delete_central
+      region: central
+      type: qos_dscp_marking_rule
+      params:
+        qos_policy: preparation@policy1@id
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            id: invalid-id
+- task_set_id: qos-policy-delete
+  depend: [preparation]
+  tasks:
+    - task_id: policy1_delete
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy1@id
+        method: delete
+    - task_id: policy2_delete
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy2@id
+        method: delete
+    - task_id: policy3_delete
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy3@id
+        method: delete
+    - task_id: policy4_delete
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy4@id
+        method: delete
+    - task_id: policy5_delete
+      region: central
+      type: qos_policy
+      action:
+        target: preparation@policy5@id
+        method: delete
+- task_set_id: wait_for_policy_delete
+  tasks:
+    - task_id: check_for_policy_delete_central
+      region: central
+      type: qos_policy
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            name: invalid-name
+    - task_id: check_for_policy_delete_region
+      region: region1
+      type: qos_policy
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            name: invalid-name
+- task_set_id: delete_vm
+  depend: [create_vm]
+  tasks:
+    - task_id: delete_vm1
+      region: region1
+      type: server
+      action:
+        target: create_vm@vm1@id
+        method: delete
+    - task_id: delete_vm2
+      region: region1
+      type: server
+      action:
+        target: create_vm@vm2@id
+        method: delete
+    - task_id: delete_vm3
+      region: region1
+      type: server
+      action:
+        target: create_vm@vm3@id
+        method: delete
+    - task_id: delete_vm4
+      region: region1
+      type: server
+      action:
+        target: create_vm@vm4@id
+        method: delete
+- task_set_id: wait_for_vm_delete
+  tasks:
+    - task_id: check_for_vm_delete
+      region: region1
+      type: server
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+            name: invalid-name
+- task_set_id: delete_net
+  depend: [preparation]
+  tasks:
+    - task_id: delete_port1
+      region: central
+      type: port
+      action:
+        target: preparation@port1@id
+        method: delete
+    - task_id: delete_port2
+      region: central
+      type: port
+      action:
+        target: preparation@port2@id
+        method: delete
+    - task_id: delete_port3
+      region: central
+      type: port
+      action:
+        target: preparation@port3@id
+        method: delete
+    - task_id: delete_port4
+      region: central
+      type: port
+      action:
+        target: preparation@port4@id
+        method: delete
+    - task_id: delete_subnet1
+      region: central
+      type: subnet
+      depend: [delete_port1]
+      action:
+        target: preparation@subnet1@id
+        method: delete
+        retries: 3
+    - task_id: delete_subnet2
+      region: central
+      type: subnet
+      depend: [delete_port2]
+      action:
+        target: preparation@subnet2@id
+        method: delete
+        retries: 3
+    - task_id: delete_subnet3
+      region: central
+      type: subnet
+      depend: [delete_port3]
+      action:
+        target: preparation@subnet3@id
+        method: delete
+        retries: 3
+    - task_id: delete_subnet4
+      region: central
+      type: subnet
+      depend: [delete_port4]
+      action:
+        target: preparation@subnet4@id
+        method: delete
+        retries: 3
+    - task_id: delete_net1
+      region: central
+      type: network
+      depend: [delete_subnet1]
+      action:
+        target: preparation@net1@id
+        method: delete
+    - task_id: delete_net2
+      region: central
+      type: network
+      depend: [delete_subnet2]
+      action:
+        target: preparation@net2@id
+        method: delete
+    - task_id: delete_net3
+      region: central
+      type: network
+      depend: [delete_subnet3]
+      action:
+        target: preparation@net3@id
+        method: delete
+    - task_id: delete_net4
+      region: central
+      type: network
+      depend: [delete_subnet4]
+      action:
+        target: preparation@net4@id
+        method: delete
+    - task_id: delete_net5
+      region: central
+      type: network
+      action:
+        target: preparation@net5@id
+        method: delete
+- task_set_id: check_net_delete
+  tasks:
+    - task_id: check_net_delete_job
+      region: region1
+      type: network
+      validate:
+        predicate: all
+        condition:
+          - name: invalid-name
+    - task_id: check-jobs
+      region: central
+      type: job
+      validate:
+        predicate: all
+        retries: 10
+        condition:
+          - status: SUCCESS

tricircle/tempestplugin/smoke_test.sh

@@ -25,3 +25,8 @@ fi
 #if [ $? != 0 ]; then
 #    die $LINENO "Smoke test fails, error in service function chain test"
 #fi
+echo "Start to run qos policy function test"
+python run_yaml_test.py qos_policy_rule_test.yaml "$OS_AUTH_URL" "$OS_TENANT_NAME" "$OS_USERNAME" "$OS_PASSWORD"
+if [ $? != 0 ]; then
+    die $LINENO "Smoke test fails, error in service function chain test"
+fi

tricircle/tempestplugin/task_runner.py

@@ -77,12 +77,15 @@ class SDKRunner(object):
     serv_reslist_map = {
         'network_sdk': ['network', 'subnet', 'port', 'router', 'fip', 'trunk',
                         'flow_classifier', 'port_pair', 'port_pair_group',
-                        'port_chain'],
+                        'port_chain', 'qos_policy', 'qos_bandwidth_limit_rule',
+                        'qos_dscp_marking_rule', 'qos_minimum_bandwidth_rule'],
         'compute': ['server'],
         'image': ['image'],
         'tricircle_sdk': ['job']}
     res_alias_map = {
         'fip': 'ip'}
+    type_plural_map = {
+        'qos_policy': 'qos_policie'}
 
     def __init__(self, auth_url, project, username, password):
         self.res_serv_map = {}
@@ -133,6 +136,7 @@ class SDKRunner(object):
         serv = self.res_serv_map[_type]
         _type = self.res_alias_map.get(_type, _type)
         proxy = getattr(conn, serv)
+        _type = self.type_plural_map.get(_type, _type)
         _list = list(getattr(proxy, '%ss' % _type)(**params))
         if get_one:
             return _list[0]

tricircle/tests/unit/network/test_central_plugin.py

@@ -38,6 +38,9 @@ from neutron.db import ipam_pluggable_backend
 from neutron.db import l3_db
 from neutron.db import models_v2
 from neutron.db import rbac_db_models as rbac_db
+from neutron.services.qos.drivers import manager as q_manager
+
+from neutron.plugins.ml2 import managers as n_managers
 
 from neutron.ipam import driver
 from neutron.ipam import exceptions as ipam_exc
@@ -60,12 +63,15 @@ import tricircle.db.api as db_api
 from tricircle.db import core
 from tricircle.db import models
 import tricircle.network.central_plugin as plugin
+from tricircle.network import central_qos_plugin
 from tricircle.network.drivers import type_flat
 from tricircle.network.drivers import type_local
 from tricircle.network.drivers import type_vlan
 from tricircle.network.drivers import type_vxlan
 from tricircle.network import helper
 from tricircle.network import managers
+from tricircle.network import qos_driver
+from tricircle.tests.unit.network import test_qos
 from tricircle.tests.unit.network import test_security_groups
 import tricircle.tests.unit.utils as test_utils
 from tricircle.xjob import xmanager
@@ -84,18 +90,24 @@ TOP_SEGMENTS = _resource_store.TOP_NETWORKSEGMENTS
 TOP_FLOATINGIPS = _resource_store.TOP_FLOATINGIPS
 TOP_SGS = _resource_store.TOP_SECURITYGROUPS
 TOP_SG_RULES = _resource_store.TOP_SECURITYGROUPRULES
+TOP_POLICIES = _resource_store.TOP_QOS_POLICIES
+TOP_POLICY_RULES = _resource_store.TOP_QOS_BANDWIDTH_LIMIT_RULES
 BOTTOM1_NETS = _resource_store.BOTTOM1_NETWORKS
 BOTTOM1_SUBNETS = _resource_store.BOTTOM1_SUBNETS
 BOTTOM1_PORTS = _resource_store.BOTTOM1_PORTS
 BOTTOM1_SGS = _resource_store.BOTTOM1_SECURITYGROUPS
 BOTTOM1_FIPS = _resource_store.BOTTOM1_FLOATINGIPS
 BOTTOM1_ROUTERS = _resource_store.BOTTOM1_ROUTERS
+BOTTOM1_POLICIES = _resource_store.BOTTOM1_QOS_POLICIES
+BOTTOM1_POLICY_RULES = _resource_store.BOTTOM1_QOS_BANDWIDTH_LIMIT_RULES
 BOTTOM2_NETS = _resource_store.BOTTOM2_NETWORKS
 BOTTOM2_SUBNETS = _resource_store.BOTTOM2_SUBNETS
 BOTTOM2_PORTS = _resource_store.BOTTOM2_PORTS
 BOTTOM2_SGS = _resource_store.BOTTOM2_SECURITYGROUPS
 BOTTOM2_FIPS = _resource_store.BOTTOM2_FLOATINGIPS
 BOTTOM2_ROUTERS = _resource_store.BOTTOM2_ROUTERS
+BOTTOM2_POLICIES = _resource_store.BOTTOM2_QOS_POLICIES
+BOTTOM2_POLICY_RULES = _resource_store.BOTTOM2_QOS_BANDWIDTH_LIMIT_RULES
 TEST_TENANT_ID = test_utils.TEST_TENANT_ID
 FakeNeutronContext = test_utils.FakeNeutronContext
 
@@ -271,7 +283,9 @@ class FakeClient(test_utils.FakeClient):
             if 'gateway_ip' not in body[_type]:
                 cidr = body[_type]['cidr']
                 body[_type]['gateway_ip'] = cidr[:cidr.rindex('.')] + '.1'
-        if 'id' not in body[_type]:
+        if _type == 'qos_policy':
+            body['policy']['id'] = uuidutils.generate_uuid()
+        elif 'id' not in body[_type]:
             body[_type]['id'] = uuidutils.generate_uuid()
         return super(FakeClient, self).create_resources(_type, ctx, body)
 
@@ -419,7 +433,7 @@ class FakeClient(test_utils.FakeClient):
         elif action == 'remove_interface':
             return self.remove_interface_routers(ctx, *args, **kwargs)
 
-    def _is_bridge_network_attached():
+    def _is_bridge_network_attached(self):
         pass
 
     def create_floatingips(self, ctx, body):
@@ -484,7 +498,86 @@ class FakeClient(test_utils.FakeClient):
         # group
         return copy.deepcopy(sg)
 
-    def get_security_group(self, context, _id, fields=None, tenant_id=None):
+    def get_security_group(self, ctx, _id, fields=None, tenant_id=None):
+        pass
+
+    def get_qos_policies(self, ctx, policy_id):
+        rules = {'rules': []}
+        rule_list = \
+            self._res_map[self.region_name]['qos_bandwidth_limit_rules']
+        for rule in rule_list:
+            if rule['qos_policy_id'] == policy_id:
+                rules['rules'].append(rule)
+
+        res_list = self._res_map[self.region_name]['qos_policy']
+        for policy in res_list:
+            if policy['id'] == policy_id:
+                policy['rules'] = rules['rules']
+                return policy
+
+    def update_qos_policies(self, ctx, policy_id, body):
+        self.update_resources('policy', ctx, policy_id, body)
+
+    def delete_qos_policies(self, ctx, policy_id):
+        self.delete_resources('policy', ctx, policy_id)
+
+    def list_bandwidth_limit_rules(self, ctx, filters):
+        policy_id = filters[0].get("value")
+        if self.region_name == 'top':
+            res_list = \
+                self._res_map[self.region_name]['qos_bandwidth_limit_rules']
+        else:
+            res_list = self._res_map[self.region_name]['qos_policy']
+            for policy in res_list:
+                if policy['id'] == policy_id:
+                    res_list = policy.get('rules', [])
+
+        ret_rules = []
+        for rule in res_list:
+            if rule['qos_policy_id'] == policy_id:
+                ret_rules.append(rule)
+
+        return ret_rules
+
+    def list_dscp_marking_rules(self, ctx, filters):
+        return []
+
+    def list_minimum_bandwidth_rules(self, ctx, filters):
+        return []
+
+    def create_bandwidth_limit_rules(self, ctx, policy_id, body):
+        res_list = self._res_map[self.region_name]['qos_policy']
+        for policy in res_list:
+            if policy['id'] == policy_id:
+                rule_id = uuidutils.generate_uuid()
+                body['bandwidth_limit_rule']['id'] = rule_id
+                body['bandwidth_limit_rule']['qos_policy_id'] = policy_id
+                policy['rules'].append(body['bandwidth_limit_rule'])
+                return body
+
+        raise q_exceptions.Conflict()
+
+    def create_dscp_marking_rules(self, ctx, policy_id, body):
+        pass
+
+    def create_minimum_bandwidth_rules(self, ctx, policy_id, body):
+        pass
+
+    def delete_bandwidth_limit_rules(self, ctx, combined_id):
+        (rule_id, policy_id) = combined_id.split('#')
+        res_list = self._res_map[self.region_name]['qos_policy']
+        for policy in res_list:
+            if policy['id'] == policy_id:
+                for rule in policy['rules']:
+                    if rule['id'] == rule_id:
+                        policy['rules'].remove(rule)
+                        return
+        raise q_exceptions.Conflict()
+
+    def delete_dscp_marking_rules(self, ctx, combined_id):
+        pass
+
+    def delete_minimum_bandwidth_rules(self, ctx, combined_id):
         pass
 
     def list_security_groups(self, ctx, sg_filters):
@@ -585,6 +678,26 @@ class FakeBaseRPCAPI(object):
     def setup_shadow_ports(self, ctxt, project_id, pod_id, net_id):
         pass
 
+    def create_qos_policy(self, ctxt, project_id, policy_id, pod_id,
+                          res_type, res_id=None):
+        combine_id = '%s#%s#%s#%s' % (pod_id, policy_id, res_type, res_id)
+        self.xmanager.create_qos_policy(
+            ctxt, payload={constants.JT_QOS_CREATE: combine_id})
+
+    def update_qos_policy(self, ctxt, project_id, policy_id, pod_id):
+        combine_id = '%s#%s' % (pod_id, policy_id)
+        self.xmanager.update_qos_policy(
+            ctxt, payload={constants.JT_QOS_UPDATE: combine_id})
+
+    def delete_qos_policy(self, ctxt, project_id, policy_id, pod_id):
+        combine_id = '%s#%s' % (pod_id, policy_id)
+        self.xmanager.delete_qos_policy(
+            ctxt, payload={constants.JT_QOS_DELETE: combine_id})
+
+    def sync_qos_policy_rules(self, ctxt, project_id, policy_id):
+        self.xmanager.sync_qos_policy_rules(
+            ctxt, payload={constants.JT_SYNC_QOS_RULE: policy_id})
+
 
 class FakeRPCAPI(FakeBaseRPCAPI):
     def __init__(self, fake_plugin):
@@ -659,12 +772,46 @@ class FakeTypeManager(managers.TricircleTypeManager):
                 break
 
 
-class FakePlugin(plugin.TricirclePlugin):
+class FakeExtensionManager(n_managers.ExtensionManager):
+    def __init__(self):
+        super(FakeExtensionManager, self).__init__()
+
+
+class FakeTricircleQoSDriver(qos_driver.TricircleQoSDriver):
+    def __init__(self, name, vif_types, vnic_types,
+                 supported_rules,
+                 requires_rpc_notifications):
+        super(FakeTricircleQoSDriver, self).__init__(
+            name, vif_types, vnic_types, supported_rules,
+            requires_rpc_notifications)
+        self.xjob_handler = FakeRPCAPI(self)
+
+    @staticmethod
+    def create():
+        return FakeTricircleQoSDriver(
+            name='tricircle',
+            vif_types=qos_driver.VIF_TYPES,
+            vnic_types=portbindings.VNIC_TYPES,
+            supported_rules=qos_driver.SUPPORTED_RULES,
+            requires_rpc_notifications=False)
+
+
+class FakeQosServiceDriverManager(q_manager.QosServiceDriverManager):
+    def __init__(self):
+        self._drivers = [FakeTricircleQoSDriver.create()]
+        self.rpc_notifications_required = False
+
+
+class FakePlugin(plugin.TricirclePlugin,
+                 central_qos_plugin.TricircleQosPlugin):
     def __init__(self):
         self.set_ipam_backend()
         self.helper = FakeHelper(self)
         self.xjob_handler = FakeRPCAPI(self)
         self.type_manager = FakeTypeManager()
+        self.extension_manager = FakeExtensionManager()
+        self.extension_manager.initialize()
+        self.driver_manager = FakeQosServiceDriverManager()
 
     def _get_client(self, region_name):
         return FakeClient(region_name)
@@ -809,7 +956,8 @@ class FakeTrunkPlugin(object):
 
 
 class PluginTest(unittest.TestCase,
-                 test_security_groups.TricircleSecurityGroupTestMixin):
+                 test_security_groups.TricircleSecurityGroupTestMixin,
+                 test_qos.TricircleQosTestMixin):
     def setUp(self):
         core.initialize()
         core.ModelBase.metadata.create_all(core.get_engine())
@@ -909,10 +1057,14 @@ class PluginTest(unittest.TestCase,
         port2 = fake_plugin.get_port(neutron_context, 'top_id_2')
         fake_plugin.get_port(neutron_context, 'top_id_3')
 
-        self.assertEqual({'id': 'top_id_1', 'name': 'bottom'}, port1)
-        self.assertEqual({'id': 'top_id_2', 'name': 'bottom'}, port2)
+        self.assertEqual({'id': 'top_id_1', 'name': 'bottom',
+                          'qos_policy_id': None}, port1)
+        self.assertEqual({'id': 'top_id_2', 'name': 'bottom',
+                          'qos_policy_id': None}, port2)
         calls = [mock.call(neutron_context, 'top_id_0', None),
-                 mock.call(neutron_context, 'top_id_3', None)]
+                 mock.call().__setitem__('qos_policy_id', None),
+                 mock.call(neutron_context, 'top_id_3', None),
+                 mock.call().__setitem__('qos_policy_id', None)]
         mock_plugin_method.assert_has_calls(calls)
 
     @patch.object(context, 'get_context_from_neutron_context',
@@ -934,11 +1086,15 @@ class PluginTest(unittest.TestCase,
                                        marker=ports3[-1]['id'])
         ports = []
         expected_ports = [{'id': 'top_id_0', 'name': 'top',
+                           'qos_policy_id': None,
                            'fixed_ips': [{'subnet_id': 'top_subnet_id',
                                           'ip_address': '10.0.0.1'}]},
-                          {'id': 'top_id_1', 'name': 'bottom'},
-                          {'id': 'top_id_2', 'name': 'bottom'},
-                          {'id': 'top_id_3', 'name': 'top'}]
+                          {'id': 'top_id_1', 'name': 'bottom',
+                           'qos_policy_id': None},
+                          {'id': 'top_id_2', 'name': 'bottom',
+                           'qos_policy_id': None},
+                          {'id': 'top_id_3', 'name': 'top',
+                           'qos_policy_id': None}]
         for _ports in (ports1, ports2, ports3, ports4):
             ports.extend(_ports)
         six.assertCountEqual(self, expected_ports, ports)
@@ -963,9 +1119,11 @@ class PluginTest(unittest.TestCase,
         ports3 = fake_plugin.get_ports(neutron_context,
                                        filters={'id': ['top_id_4']})
         self.assertEqual([{'id': 'top_id_0', 'name': 'top',
+                           'qos_policy_id': None,
                            'fixed_ips': [{'subnet_id': 'top_subnet_id',
                                           'ip_address': '10.0.0.1'}]}], ports1)
-        self.assertEqual([{'id': 'top_id_1', 'name': 'bottom'}], ports2)
+        self.assertEqual([{'id': 'top_id_1', 'name': 'bottom',
+                           'qos_policy_id': None}], ports2)
         self.assertEqual([], ports3)
 
         TOP_ROUTERS.append({'id': 'router_id'})
@@ -990,9 +1148,9 @@ class PluginTest(unittest.TestCase,
         ports = fake_plugin.get_ports(neutron_context,
                                       filters={'device_id': ['router_id']})
         expected = [{'id': 'top_id_1', 'name': 'bottom',
-                     'device_id': 'router_id'},
+                     'qos_policy_id': None, 'device_id': 'router_id'},
                     {'id': 'top_id_2', 'name': 'bottom',
-                     'device_id': 'router_id'}]
+                     'qos_policy_id': None, 'device_id': 'router_id'}]
         six.assertCountEqual(self, expected, ports)
 
     @patch.object(context, 'get_context_from_neutron_context')
@@ -3372,6 +3530,97 @@ class PluginTest(unittest.TestCase,
                                      'pod_id_1', TOP_SGS,
                                      TOP_SG_RULES, BOTTOM1_SGS)
 
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_create_policy(self, mock_context):
+        fake_plugin = FakePlugin()
+        q_ctx = FakeNeutronContext()
+        t_ctx = context.get_db_context()
+        mock_context.return_value = t_ctx
+
+        self._test_create_policy(fake_plugin, q_ctx, t_ctx)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_update_policy(self, mock_context):
+        self._basic_pod_route_setup()
+        fake_plugin = FakePlugin()
+        q_ctx = FakeNeutronContext()
+        t_ctx = context.get_db_context()
+        mock_context.return_value = t_ctx
+
+        self._test_update_policy(fake_plugin, q_ctx, t_ctx, 'pod_id_1',
+                                 BOTTOM1_POLICIES)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_delete_policy(self, mock_context):
+        self._basic_pod_route_setup()
+        fake_plugin = FakePlugin()
+        q_ctx = FakeNeutronContext()
+        t_ctx = context.get_db_context()
+        mock_context.return_value = t_ctx
+
+        self._test_delete_policy(fake_plugin, q_ctx, t_ctx, 'pod_id_1',
+                                 BOTTOM1_POLICIES)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_create_policy_rule(self, mock_context):
+        self._basic_pod_route_setup()
+        fake_plugin = FakePlugin()
+        q_ctx = FakeNeutronContext()
+        t_ctx = context.get_db_context()
+        mock_context.return_value = t_ctx
+
+        self._test_create_policy_rule(fake_plugin, q_ctx, t_ctx, 'pod_id_1',
+                                      BOTTOM1_POLICIES)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_delete_policy_rule(self, mock_context):
+        self._basic_pod_route_setup()
+        fake_plugin = FakePlugin()
+        q_ctx = FakeNeutronContext()
+        t_ctx = context.get_db_context()
+        mock_context.return_value = t_ctx
+
+        self._test_delete_policy_rule(fake_plugin, q_ctx, t_ctx, 'pod_id_1',
+                                      BOTTOM1_POLICIES)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_update_network_with_qos_policy(self, mock_context):
+        self._basic_pod_route_setup()
+        t_ctx = context.get_db_context()
+        fake_plugin = FakePlugin()
+        fake_client = FakeClient('pod_1')
+        q_ctx = FakeNeutronContext()
+        mock_context.return_value = t_ctx
+
+        tenant_id = TEST_TENANT_ID
+        net_id, _, _, _ = \
+            self._prepare_network_subnet(tenant_id, t_ctx, 'pod_1', 1)
+
+        self._test_update_network_with_qos_policy(fake_plugin, fake_client,
+                                                  q_ctx, t_ctx, 'pod_id_1',
+                                                  net_id, BOTTOM1_POLICIES)
+
+    @patch.object(context, 'get_context_from_neutron_context')
+    def test_update_port_with_qos_policy(self, mock_context):
+        project_id = TEST_TENANT_ID
+        self._basic_pod_route_setup()
+        q_ctx = FakeNeutronContext()
+        fake_client = FakeClient('pod_1')
+        t_ctx = context.get_db_context()
+        fake_plugin = FakePlugin()
+        mock_context.return_value = t_ctx
+        (t_net_id, t_subnet_id,
+         b_net_id, b_subnet_id) = self._prepare_network_subnet(
+            project_id, t_ctx, 'pod_1', 1)
+        t_port_id, b_port_id = self._prepare_port_test(
+            project_id, t_ctx, 'pod_1', 1, t_net_id, b_net_id,
+            t_subnet_id, b_subnet_id)
+
+        self._test_update_port_with_qos_policy(fake_plugin, fake_client,
+                                               q_ctx, t_ctx,
+                                               'pod_id_1', t_port_id,
+                                               b_port_id, BOTTOM1_POLICIES)
+
     @patch.object(FakeBaseRPCAPI, 'setup_shadow_ports')
     @patch.object(FakeClient, 'update_ports')
     @patch.object(context, 'get_context_from_neutron_context')

tricircle/tests/unit/network/test_qos.py

@@ -0,0 +1,249 @@
+# Copyright 2015 Huawei Technologies Co., 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 neutron.objects.qos import rule
+from oslo_utils import uuidutils
+
+from tricircle.common import constants
+from tricircle.db import api as db_api
+
+
+class TricircleQosTestMixin(object):
+    def _test_create_policy(self, plugin, q_ctx, t_ctx):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id
+            }
+        }
+
+        res = plugin.create_policy(q_ctx, t_policy)
+        res1 = plugin.get_policy(q_ctx, res['id'])
+
+        self.assertEqual('test_qos', res['name'])
+        self.assertEqual(res1['id'], res['id'])
+        self.assertEqual(res1['name'], res['name'])
+        self.assertEqual(res['description'], res['description'])
+
+    def _test_update_policy(self, plugin, q_ctx, t_ctx,
+                            pod_id, bottom_policy):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id
+            }
+        }
+
+        res = plugin.create_policy(q_ctx, t_policy)
+
+        updated_qos = {
+            'policy': {
+                'name': 'test_updated_qos'
+            }
+        }
+
+        updated_res = plugin.update_policy(q_ctx, res['id'], updated_qos)
+        self.assertEqual(res['id'], updated_res['id'])
+        self.assertEqual('test_updated_qos', updated_res['name'])
+
+        b_policy_id = uuidutils.generate_uuid()
+        b_policy = {
+            'id': b_policy_id, 'name': b_policy_id, 'description': '',
+            'tenant_id': project_id
+        }
+        bottom_policy.append(b_policy)
+        db_api.create_resource_mapping(t_ctx, res['id'], b_policy_id,
+                                       pod_id, project_id, constants.RT_QOS)
+
+        updated_qos = {
+            'policy': {
+                'name': 'test_policy'
+            }
+        }
+
+        updated_res = plugin.update_policy(q_ctx, res['id'], updated_qos)
+        self.assertEqual('test_policy', updated_res['name'])
+        self.assertEqual('test_policy', bottom_policy[0]['name'])
+
+    def _test_delete_policy(self, plugin, q_ctx,
+                            t_ctx, pod_id, bottom_policy):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id
+            }
+        }
+
+        res = plugin.create_policy(q_ctx, t_policy)
+        b_policy_id = uuidutils.generate_uuid()
+        b_policy = {
+            'id': b_policy_id, 'name': b_policy_id, 'description': '',
+            'tenant_id': project_id
+        }
+        bottom_policy.append(b_policy)
+        db_api.create_resource_mapping(t_ctx, res['id'], b_policy_id,
+                                       pod_id, project_id, constants.RT_QOS)
+
+        self.assertEqual(1, len(bottom_policy))
+        plugin.delete_policy(q_ctx, res['id'])
+        self.assertEqual(0, len(bottom_policy))
+
+    def _test_create_policy_rule(self, plugin, q_ctx,
+                                 t_ctx, pod_id, bottom_policy):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id
+            }
+        }
+
+        res = plugin.create_policy(q_ctx, t_policy)
+
+        rule_data = {
+            "bandwidth_limit_rule": {
+                "max_kbps": "10000"
+            }
+        }
+
+        t_rule = plugin.create_policy_rule(
+            q_ctx, rule.QosBandwidthLimitRule, res['id'], rule_data)
+        res1 = plugin.get_policy(q_ctx, res['id'])
+
+        self.assertEqual(1, len(res1['rules']))
+        self.assertEqual(t_rule['id'], res1['rules'][0]['id'])
+
+        b_policy_id = uuidutils.generate_uuid()
+        b_policy = {'id': b_policy_id, 'name': b_policy_id, 'description': '',
+                    'tenant_id': project_id, 'rules': []}
+        bottom_policy.append(b_policy)
+        db_api.create_resource_mapping(t_ctx, res['id'], b_policy_id,
+                                       pod_id, project_id, constants.RT_QOS)
+
+        rule_data = {
+            "bandwidth_limit_rule": {
+                "max_kbps": "10000",
+                "max_burst_kbps": "20000"
+            }
+        }
+
+        t_rule = plugin.create_policy_rule(
+            q_ctx, rule.QosBandwidthLimitRule, res['id'], rule_data)
+
+        self.assertEqual(2, len(bottom_policy[0]['rules']))
+        b_rule = bottom_policy[0]['rules'][0]
+        self.assertEqual(b_policy_id, b_rule['qos_policy_id'])
+        b_rule = bottom_policy[0]['rules'][1]
+        self.assertEqual(b_policy_id, b_rule['qos_policy_id'])
+
+    def _test_delete_policy_rule(self, plugin, q_ctx,
+                                 t_ctx, pod_id, bottom_policy):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id
+            }
+        }
+
+        res = plugin.create_policy(q_ctx, t_policy)
+
+        b_policy_id = uuidutils.generate_uuid()
+        b_policy = {
+            'id': b_policy_id, 'name': b_policy_id, 'description': '',
+            'tenant_id': project_id, 'rules': []
+        }
+        bottom_policy.append(b_policy)
+        db_api.create_resource_mapping(t_ctx, res['id'], b_policy_id,
+                                       pod_id, project_id, constants.RT_QOS)
+
+        rule_data = {
+            "bandwidth_limit_rule": {
+                "max_kbps": "10000"
+            }
+        }
+
+        res1 = plugin.create_policy_rule(
+            q_ctx, rule.QosBandwidthLimitRule, res['id'], rule_data)
+
+        self.assertEqual(1, len(bottom_policy[0]['rules']))
+        b_rule = bottom_policy[0]['rules'][0]
+        self.assertEqual(b_policy_id, b_rule['qos_policy_id'])
+
+        plugin.delete_policy_rule(
+            q_ctx, rule.QosBandwidthLimitRule, res1['id'], res['id'])
+        self.assertEqual(0, len(bottom_policy[0]['rules']))
+
+    @staticmethod
+    def _create_policy_in_top(self, plugin, q_ctx, t_ctx,
+                              pod_id, bottom_policy):
+        project_id = 'test_prject_id'
+        t_policy = {
+            'policy': {
+                'name': 'test_qos',
+                'description': 'This policy limits the ports to 10Mbit max.',
+                'project_id': project_id,
+            }
+        }
+
+        return plugin.create_policy(q_ctx, t_policy)
+
+    def _test_update_network_with_qos_policy(self, plugin, client, q_ctx,
+                                             t_ctx, pod_id, t_net_id,
+                                             bottom_policy):
+        res = \
+            self._create_policy_in_top(self, plugin, q_ctx, t_ctx,
+                                       pod_id, bottom_policy)
+
+        update_body = {
+            'network': {
+                'qos_policy_id': res['id']}
+        }
+        top_net = plugin.update_network(q_ctx, t_net_id, update_body)
+        self.assertEqual(top_net['qos_policy_id'], res['id'])
+
+        route_res = \
+            db_api.get_bottom_mappings_by_top_id(t_ctx, res['id'],
+                                                 constants.RT_QOS)
+        bottom_net = client.get_networks(q_ctx, t_net_id)
+        self.assertEqual(bottom_net['qos_policy_id'], route_res[0][1])
+
+    def _test_update_port_with_qos_policy(self, plugin, client, q_ctx,
+                                          t_ctx, pod_id, t_port_id,
+                                          b_port_id, bottom_policy):
+        res = \
+            self._create_policy_in_top(self, plugin, q_ctx, t_ctx,
+                                       pod_id, bottom_policy)
+
+        update_body = {
+            'port': {
+                'qos_policy_id': res['id']}
+        }
+        top_port = plugin.update_port(q_ctx, t_port_id, update_body)
+        self.assertEqual(top_port['qos_policy_id'], res['id'])
+
+        route_res = \
+            db_api.get_bottom_mappings_by_top_id(t_ctx, res['id'],
+                                                 constants.RT_QOS)
+        bottom_port = client.get_ports(q_ctx, b_port_id)
+        self.assertEqual(bottom_port['qos_policy_id'], route_res[0][1])

tricircle/tests/unit/utils.py

@@ -55,7 +55,10 @@ class ResourceStore(object):
                       ('sfc_port_pairs', constants.RT_PORT_PAIR),
                       ('sfc_port_pair_groups', constants.RT_PORT_PAIR_GROUP),
                       ('sfc_port_chains', constants.RT_PORT_CHAIN),
-                      ('sfc_flow_classifiers', constants.RT_FLOW_CLASSIFIER)]
+                      ('sfc_flow_classifiers', constants.RT_FLOW_CLASSIFIER),
+                      ('qos_policies', constants.RT_QOS),
+                      ('qos_bandwidth_limit_rules',
+                       'qos_bandwidth_limit_rules')]
 
     def __init__(self):
         self.store_list = []
@@ -556,9 +559,13 @@ class FakeClient(object):
 
     def create_resources(self, _type, ctx, body):
         res_list = self._res_map[self.region_name][_type]
+        if _type == 'qos_policy':
+            _type = 'policy'
         res = dict(body[_type])
         if 'id' not in res:
             res['id'] = uuidutils.generate_uuid()
+        if _type == 'policy' and 'rules' not in res:
+            res['rules'] = []
         res_list.append(res)
         return res
 
@@ -586,6 +593,8 @@ class FakeClient(object):
         return None
 
     def delete_resources(self, _type, ctx, _id):
+        if _type is 'policy':
+            _type = 'qos_policy'
         index = -1
         res_list = self._res_map[self.region_name][_type]
         for i, res in enumerate(res_list):
@@ -595,6 +604,9 @@ class FakeClient(object):
             del res_list[index]
 
     def update_resources(self, _type, ctx, _id, body):
+        if _type == 'policy':
+            res_list = self._res_map[self.region_name]['qos_policy']
+        else:
             res_list = self._res_map[self.region_name][_type]
         updated = False
         for res in res_list:

tricircle/xjob/xmanager.py

@@ -159,7 +159,12 @@ class XManager(PeriodicTasks):
             constants.JT_NETWORK_UPDATE: self.update_network,
             constants.JT_SUBNET_UPDATE: self.update_subnet,
             constants.JT_SHADOW_PORT_SETUP: self.setup_shadow_ports,
-            constants.JT_TRUNK_SYNC: self.sync_trunk}
+            constants.JT_TRUNK_SYNC: self.sync_trunk,
+            constants.JT_QOS_CREATE: self.create_qos_policy,
+            constants.JT_QOS_UPDATE: self.update_qos_policy,
+            constants.JT_QOS_DELETE: self.delete_qos_policy,
+            constants.JT_SYNC_QOS_RULE: self.sync_qos_policy_rules
+        }
         self.helper = helper.NetworkHelper()
         self.xjob_handler = xrpcapi.XJobAPI()
         super(XManager, self).__init__()
@@ -898,8 +903,9 @@ class XManager(PeriodicTasks):
                 ctx, t_network_id, constants.RT_NETWORK)
             b_pods = [mapping[0] for mapping in mappings]
             for b_pod in b_pods:
-                self.xjob_handler.update_network(ctx, project_id,
-                                                 t_network_id, b_pod['pod_id'])
+                self.xjob_handler.update_network(
+                    ctx, project_id, t_network_id,
+                    b_pod['pod_id'])
             return
 
         b_pod = db_api.get_pod(ctx, b_pod_id)
@@ -918,6 +924,9 @@ class XManager(PeriodicTasks):
             }
         }
 
+        if not t_network.get('qos_policy_id', None):
+            body['network']['qos_policy_id'] = None
+
         try:
             b_client.update_networks(ctx, b_network_id, body)
         except q_cli_exceptions.NotFound:
@@ -1527,3 +1536,287 @@ class XManager(PeriodicTasks):
             fc_id=b_fc_ids[0])
 
         self.xjob_handler.recycle_resources(ctx, t_pc['project_id'])
+
+    @_job_handle(constants.JT_QOS_CREATE)
+    def create_qos_policy(self, ctx, payload):
+        (b_pod_id, t_policy_id, res_type, res_id) = payload[
+            constants.JT_QOS_CREATE].split('#')
+
+        t_client = self._get_client()
+        t_policy = t_client.get_qos_policies(ctx, t_policy_id)
+
+        if not t_policy:
+            # we just end this job if top policy no longer exists
+            return
+
+        project_id = t_policy['tenant_id']
+
+        if b_pod_id == constants.POD_NOT_SPECIFIED:
+            mappings = db_api.get_bottom_mappings_by_top_id(
+                ctx, res_id, res_type)
+            for b_pod, _ in mappings:
+                self.xjob_handler.create_qos_policy(ctx, project_id,
+                                                    t_policy_id,
+                                                    b_pod['pod_id'],
+                                                    res_type,
+                                                    res_id)
+            return
+
+        b_pod = db_api.get_pod(ctx, b_pod_id)
+        b_region_name = b_pod['region_name']
+        b_client = self._get_client(region_name=b_region_name)
+
+        body = {
+            'policy': {
+                'description': t_policy.get('description', ''),
+                'tenant_id': t_policy.get('tenant_id', ''),
+                'project_id': t_policy.get('project_id', ''),
+                'shared': t_policy.get('shared', False),
+                'name': t_policy.get('name', '')
+            }
+        }
+
+        try:
+            _, b_policy_id = self.helper.prepare_bottom_element(
+                ctx, project_id, b_pod, t_policy, constants.RT_QOS, body)
+            if res_id:
+                if res_type == constants.RT_NETWORK:
+                    body = {
+                        "network": {
+                            "qos_policy_id": b_policy_id
+                        }
+                    }
+                    b_client.update_networks(ctx, res_id, body)
+                if res_type == constants.RT_PORT:
+                    body = {
+                        "port": {
+                            "qos_policy_id": b_policy_id
+                        }
+                    }
+                    b_client.update_ports(ctx, res_id, body)
+            if t_policy['rules']:
+                self.xjob_handler.sync_qos_policy_rules(
+                    ctx, project_id, t_policy_id)
+        except q_cli_exceptions.NotFound:
+            LOG.error('qos policy: %(policy_id)s not found,'
+                      'pod name: %(name)s',
+                      {'policy_id': t_policy_id, 'name': b_region_name})
+
+    @_job_handle(constants.JT_QOS_UPDATE)
+    def update_qos_policy(self, ctx, payload):
+        (b_pod_id, t_policy_id) = payload[
+            constants.JT_QOS_UPDATE].split('#')
+
+        t_client = self._get_client()
+        t_policy = t_client.get_qos_policies(ctx, t_policy_id)
+        if not t_policy:
+            return
+        project_id = t_policy['tenant_id']
+
+        if b_pod_id == constants.POD_NOT_SPECIFIED:
+            mappings = db_api.get_bottom_mappings_by_top_id(
+                ctx, t_policy_id, constants.RT_QOS)
+            for b_pod, _ in mappings:
+                self.xjob_handler.update_qos_policy(ctx, project_id,
+                                                    t_policy_id,
+                                                    b_pod['pod_id'])
+            return
+
+        b_pod = db_api.get_pod(ctx, b_pod_id)
+        b_region_name = b_pod['region_name']
+        b_client = self._get_client(region_name=b_region_name)
+        b_policy_id = db_api.get_bottom_id_by_top_id_region_name(
+            ctx, t_policy_id, b_region_name, constants.RT_QOS)
+
+        if not b_policy_id:
+            return
+
+        body = {
+            'policy': {
+                'description': t_policy.get('description', ''),
+                'shared': t_policy.get('shared', ''),
+                'name': t_policy.get('name', '')
+            }
+        }
+
+        try:
+            b_client.update_qos_policies(ctx, b_policy_id, body)
+        except q_cli_exceptions.NotFound:
+            LOG.error('qos policy: %(policy_id)s not found,'
+                      'pod name: %(name)s',
+                      {'policy_id': t_policy_id, 'name': b_region_name})
+
+    @_job_handle(constants.JT_QOS_DELETE)
+    def delete_qos_policy(self, ctx, payload):
+        (b_pod_id, t_policy_id) = payload[
+            constants.JT_QOS_DELETE].split('#')
+
+        project_id = ctx.tenant_id
+        if b_pod_id == constants.POD_NOT_SPECIFIED:
+            mappings = db_api.get_bottom_mappings_by_top_id(
+                ctx, t_policy_id, constants.RT_QOS)
+            for b_pod, _ in mappings:
+                self.xjob_handler.delete_qos_policy(ctx, project_id,
+                                                    t_policy_id,
+                                                    b_pod['pod_id'])
+            return
+
+        b_pod = db_api.get_pod(ctx, b_pod_id)
+        b_region_name = b_pod['region_name']
+        b_client = self._get_client(region_name=b_region_name)
+        b_policy_id = db_api.get_bottom_id_by_top_id_region_name(
+            ctx, t_policy_id, b_region_name, constants.RT_QOS)
+
+        try:
+            b_client.delete_qos_policies(ctx, b_policy_id)
+            db_api.delete_mappings_by_bottom_id(ctx, b_policy_id)
+        except q_cli_exceptions.NotFound:
+            LOG.error('qos policy: %(policy_id)s not found,'
+                      'pod name: %(name)s',
+                      {'policy_id': t_policy_id, 'name': b_region_name})
+
+    @staticmethod
+    def _safe_create_policy_rule(
+            t_context, client, rule_type, policy_id, body):
+        try:
+            return getattr(client, 'create_%s_rules' % rule_type)(
+                t_context, policy_id, body)
+        except q_exceptions.Conflict:
+            return
+
+    @staticmethod
+    def _safe_get_policy_rule(t_context, client, rule_type, rule_id,
+                              policy_id):
+        combine_id = '%s#%s' % (rule_id, policy_id)
+        return getattr(client, 'get_%s_rules' % rule_type)(
+            t_context, combine_id)
+
+    @staticmethod
+    def _safe_delete_policy_rule(t_context, client, rule_type, rule_id,
+                                 policy_id):
+        combine_id = '%s#%s' % (rule_id, policy_id)
+        getattr(client, 'delete_%s_rules' % rule_type)(
+            t_context, combine_id)
+
+    @staticmethod
+    def _construct_bottom_bandwidth_limit_rule(t_rule):
+        return {
+            "max_kbps": t_rule["max_kbps"],
+            "max_burst_kbps": t_rule["max_burst_kbps"]
+        }
+
+    @staticmethod
+    def _construct_bottom_dscp_marking_rule(t_rule):
+        return {
+            "dscp_mark": t_rule["dscp_mark"]
+        }
+
+    @staticmethod
+    def _construct_bottom_minimum_bandwidth_rule(t_rule):
+        return {
+            "min_kbps": t_rule["min_kbps"],
+            "direction": t_rule["direction"]
+        }
+
+    def _construct_bottom_policy_rule(self, rule_type, rule_data):
+        return getattr(self, '_construct_bottom_%s_rule' % rule_type)(
+            rule_data)
+
+    @staticmethod
+    def _compare_bandwidth_limit_rule(rule1, rule2):
+        for key in ('max_kbps', 'max_burst_kbps'):
+            if rule1[key] != rule2[key]:
+                return False
+        return True
+
+    @staticmethod
+    def _compare_dscp_marking_rule(rule1, rule2):
+        for key in ('dscp_mark',):
+            if rule1[key] != rule2[key]:
+                return False
+        return True
+
+    @staticmethod
+    def _compare_minimum_bandwidth_rule(rule1, rule2):
+        for key in ('min_kbps', 'direction'):
+            if rule1[key] != rule2[key]:
+                return False
+        return True
+
+    def _compare_policy_rule(self, rule_type, rule1, rule2):
+        return getattr(self, '_compare_%s_rule' % rule_type)(rule1, rule2)
+
+    @_job_handle(constants.JT_SYNC_QOS_RULE)
+    def sync_qos_policy_rules(self, ctx, payload):
+        policy_id = payload[constants.JT_SYNC_QOS_RULE]
+        top_client = self._get_client()
+
+        bandwidth_limit_rules = \
+            top_client.list_bandwidth_limit_rules(ctx, filters=[{
+                'key': 'policy_id', 'comparator': 'eq', 'value': policy_id}])
+        dscp_marking_rules = \
+            top_client.list_dscp_marking_rules(ctx, filters=[{
+                'key': 'policy_id', 'comparator': 'eq', 'value': policy_id}])
+        minimum_bandwidth_rules = \
+            top_client.list_minimum_bandwidth_rules(ctx, filters=[{
+                'key': 'policy_id', 'comparator': 'eq', 'value': policy_id}])
+        mappings = db_api.get_bottom_mappings_by_top_id(
+            ctx, policy_id, constants.RT_QOS)
+
+        self._sync_policy_rules(
+            ctx, mappings, 'bandwidth_limit_rule', bandwidth_limit_rules)
+
+        self._sync_policy_rules(
+            ctx, mappings, 'dscp_marking_rule', dscp_marking_rules)
+
+        self._sync_policy_rules(
+            ctx, mappings, 'minimum_bandwidth_rule', minimum_bandwidth_rules)
+
+    def _sync_policy_rules(self, ctx, mappings, rule_type, rules):
+        if rule_type == 'bandwidth_limit_rule':
+            rule_types = 'bandwidth_limit_rules'
+            prefix = 'bandwidth_limit'
+        elif rule_type == 'dscp_marking_rule':
+            rule_types = 'dscp_marking_rules'
+            prefix = 'dscp_marking'
+        else:
+            rule_types = 'minimum_bandwidth_rules'
+            prefix = 'minimum_bandwidth'
+
+        new_b_rules = []
+        for t_rule in rules:
+            new_b_rules.append(
+                getattr(self, '_construct_bottom_%s' % rule_type)(t_rule))
+
+        for pod, b_policy_id in mappings:
+            client = self._get_client(pod['region_name'])
+            b_rules = getattr(client, 'list_%s' % rule_types)(
+                ctx, filters=[{'key': 'policy_id',
+                               'comparator': 'eq',
+                               'value': b_policy_id}])
+            add_rules = []
+            del_rules = []
+            match_index = set()
+            for b_rule in b_rules:
+                match = False
+                for i, rule in enumerate(new_b_rules):
+                    if getattr(self, '_compare_%s' % rule_type)(b_rule, rule):
+                        match = True
+                        match_index.add(i)
+                        break
+                if not match:
+                    del_rules.append(b_rule)
+            for i, rule in enumerate(new_b_rules):
+                if i not in match_index:
+                    add_rules.append(rule)
+
+            for del_rule in del_rules:
+                self._safe_delete_policy_rule(
+                    ctx, client, prefix, del_rule['id'],
+                    b_policy_id)
+            if add_rules:
+                for new_rule in add_rules:
+                    rule_body = {rule_type: new_rule}
+                    self._safe_create_policy_rule(
+                        ctx, client, prefix,
+                        b_policy_id, rule_body)