[stable-only] Do not fail making reservation when creating a SG

Do not fail during the creation of a security group when trying to
make a quota reservation for the security group rules. This feature
was added in [1], in order to prevent the rule quota excess during
the security group creation.

However, as reported in LP#1992161, this method can be called from
the RPC worker. If this RPC worker is spawned alone (not with the API
workers), the extensions are not loaded and the security group rule
quota resources are not created. That means the quota engine does not
have the security group rules as managed resources (in this worker).

When a new network (and the first subnet) is created, the DHCP agent
(or agents) handling this network will try to create the DHCP port.
If, as commented in the LP bug, the default security group is not
created, the RPC worker will try to create it. In this case this
patch skips the quota check.

This patch is for stable releases only. Since Xena, this check is
done using a new method called "quota_limit_check" [2]. This method
does not fail in the related case.

[1]https://review.opendev.org/q/I0a9b91b09d6260ff96fdba2f0a455de53bbc1f00
[2]https://review.opendev.org/q/Id73368576a948f78a043d7cf0be16661a65626a9

Conflicts:
      neutron/db/securitygroups_db.py

Closes-Bug: #1992161
Related-Bug: #1858680
Change-Id: I0f20b17c1b13c3cf56de70588fca4a6956d276df
(cherry picked from commit 02bdd04702)
(cherry picked from commit 90865c06af)

neutron/db/securitygroups_db.py

@@ -120,7 +120,7 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase,
             delta = delta * 2 if default_sg else delta
             reservation = quota.QUOTAS.make_reservation(
                 context, tenant_id, {'security_group_rule': delta},
-                self)
+                self, raise_exception=False)
 
             for ethertype in ext_sg.sg_supported_ethertypes:
                 if default_sg:
@@ -141,8 +141,9 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase,
                 sg.rules.append(egress_rule)
             sg.obj_reset_changes(['rules'])
 
-            quota.QUOTAS.commit_reservation(context,
-                                            reservation.reservation_id)
+            if reservation:
+                quota.QUOTAS.commit_reservation(context,
+                                                reservation.reservation_id)
 
             # fetch sg from db to load the sg rules with sg model.
             sg = sg_obj.SecurityGroup.get_object(context, id=sg.id)

neutron/quota/__init__.py

@@ -205,7 +205,8 @@ class QuotaEngine(object):
 
         return res.count(context, *args, **kwargs)
 
-    def make_reservation(self, context, tenant_id, deltas, plugin):
+    def make_reservation(self, context, tenant_id, deltas, plugin,
+                         raise_exception=True):
         # Verify that resources are managed by the quota engine
         # Ensure no value is less than zero
         unders = [key for key, val in deltas.items() if val < 0]
@@ -220,8 +221,10 @@ class QuotaEngine(object):
         unknown_resources = requested_resources - managed_resources
 
         if unknown_resources:
-            raise exceptions.QuotaResourceUnknown(
-                unknown=sorted(unknown_resources))
+            if raise_exception:
+                raise exceptions.QuotaResourceUnknown(
+                    unknown=sorted(unknown_resources))
+            return
         # FIXME(salv-orlando): There should be no reason for sending all the
         # resource in the registry to the quota driver, but as other driver
         # APIs request them, this will be sorted out with a different patch.

neutron/tests/unit/db/test_securitygroups_db.py

@@ -90,6 +90,18 @@ class SecurityGroupDbMixinTestCase(testlib_api.SqlTestCase):
                     securitygroup.SecurityGroupConflict):
                 self.mixin.create_security_group(self.ctx, secgroup)
 
+    def test_create_security_group_no_quota(self):
+        with mock.patch.object(self.mixin, '_registry_notify'):
+            self.mock_quota_make_res.return_value = None
+            self.mixin.create_security_group(self.ctx, FAKE_SECGROUP)
+            self.mock_quota_commit_res.assert_not_called()
+
+            self.mock_quota_make_res.return_value = mock.Mock(
+                reservation_id='res_id')
+            self.mixin.create_security_group(self.ctx, FAKE_SECGROUP)
+            self.mock_quota_commit_res.assert_called_once_with(self.ctx,
+                                                               'res_id')
+
     def test_delete_security_group_in_use(self):
         with mock.patch.object(self.mixin,
                                '_get_port_security_group_bindings'),\