Fix creation of trusted queues for the NVP plugin.

Currently if an admin tries to create a trusted queue, Neutron will
fail with MissingDSCPForTrusted exception. However, the NVP semantic
is exactly the opposite, namely the DSCP field must be specified for
untrusted queues and omitted trusted ones.

Fixes bug #1204256

Change-Id: I935fab45fc811a296411283a641b66c5ca96264d

neutron/plugins/nicira/dbexts/nicira_qos_db.py

@@ -22,11 +22,15 @@ from sqlalchemy.orm import exc
 from neutron.api.v2 import attributes as attr
 from neutron.db import model_base
 from neutron.db import models_v2
+from neutron.openstack.common import log
 from neutron.openstack.common import uuidutils
 from neutron.plugins.nicira.extensions import nvp_qos as ext_qos
 from neutron.plugins.nicira import nvplib
 
 
+LOG = log.getLogger(__name__)
+
+
 class QoSQueue(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
     name = sa.Column(sa.String(255))
     default = sa.Column(sa.Boolean, default=False)
@@ -267,9 +271,10 @@ class NVPQoSDbMixin(ext_qos.QueuePluginBase):
                     raise ext_qos.DefaultQueueAlreadyExists()
             else:
                 raise ext_qos.DefaultQueueCreateNotAdmin()
-        if (qos_queue.get('qos_marking') == 'trusted' and
+        if qos_queue.get('qos_marking') == 'trusted':
-            not qos_queue.get('dscp')):
+            dscp = qos_queue.pop('dscp')
-            raise ext_qos.MissingDSCPForTrusted()
+            LOG.info(_("DSCP value (%s) will be ignored with 'trusted' "
+                     "marking"), dscp)
         max = qos_queue.get('max')
         min = qos_queue.get('min')
         # Max can be None

neutron/plugins/nicira/extensions/nvp_qos.py

@@ -56,10 +56,6 @@ class QueueInvalidBandwidth(qexception.InvalidInput):
                 " integer.")
 
 
-class MissingDSCPForTrusted(qexception.InvalidInput):
-    message = _("No DSCP field needed when QoS workload marked trusted")
-
-
 class QueueNotFound(qexception.NotFound):
     message = _("Queue %(id)s does not exist")
 
@@ -91,7 +87,19 @@ def convert_to_unsigned_int_or_none_max_63(val):
         raise QueueInvalidDscp(data=val)
     return val
 
-# Attribute Map
+# As per NVP API, if a queue is trusted, DSCP must be omitted; if a queue is
+# untrusted, DSCP must be specified. Whichever default values we choose for
+# the tuple (qos_marking, dscp), there will be at least one combination of a
+# request with conflicting values: for instance, with the following default:
+#
+# qos_marking = 'untrusted', dscp = '0'
+#
+# requests with qos_marking = 'trusted' and a default dscp will fail. Since
+# it is convoluted to ask the admin to specify a None value for dscp when
+# qos_marking is 'trusted', it is best to ignore the dscp value, regardless
+# of whether it has been specified or not. This preserves the chosen default
+# and keeps backward compatibility with the API. A warning will be logged, as
+# the server is overriding a potentially conflicting request from the admin
 RESOURCE_ATTRIBUTE_MAP = {
     'qos_queues': {
         'id': {'allow_post': False, 'allow_put': False,

neutron/tests/unit/nicira/test_nicira_plugin.py

@@ -31,6 +31,7 @@ from neutron.extensions import securitygroup as secgrp
 from neutron import manager
 from neutron.openstack.common import uuidutils
 import neutron.plugins.nicira as nvp_plugin
+from neutron.plugins.nicira.dbexts import nicira_qos_db as qos_db
 from neutron.plugins.nicira.extensions import nvp_networkgw
 from neutron.plugins.nicira.extensions import nvp_qos as ext_qos
 from neutron.plugins.nicira import NeutronPlugin
@@ -666,6 +667,15 @@ class TestNiciraQoSQueue(NiciraPluginV2TestCase):
             self.assertEqual(q['qos_queue']['qos_marking'], 'untrusted')
             self.assertFalse(q['qos_queue']['default'])
 
+    def test_create_trusted_qos_queue(self):
+        with mock.patch.object(qos_db.LOG, 'info') as log:
+            with mock.patch.object(nvplib, 'do_request',
+                                   return_value={"uuid": "fake_queue"}):
+                with self.qos_queue(name='fake_lqueue', min=34, max=44,
+                                    qos_marking='trusted', default=False) as q:
+                    self.assertEqual(q['qos_queue']['dscp'], None)
+                    self.assertTrue(log.called)
+
     def test_create_qos_queue_name_exceeds_40_chars(self):
         name = 'this_is_a_queue_whose_name_is_longer_than_40_chars'
         with self.qos_queue(name=name) as queue: