Mitigate restriction for fixed ips per dhcp port

When number of fixed ips per dhcp port exceeds max_fixed_ips_per_port,
a dhcp resync will be triggered.

The bug report stated how simply this issue can be triggered.
Moreover, "max_fixed_ips_per_port" value should be used for non-DHCP
port only and DHCP port is not affected by this parameter.

Change-Id: Iaa9ed6949383ba6a7ce0b3ffd9dcced663126317
Co-authored-by: NGUYEN TUONG THANH <thanhnt@vn.fujitsu.com>
Closes-Bug: #1179713

neutron/db/ipam_backend_mixin.py

@@ -27,6 +27,7 @@ from neutron.api.v2 import attributes
 from neutron.common import constants
 from neutron.common import exceptions as n_exc
 from neutron.common import ipv6_utils
+from neutron.common import utils as common_utils
 from neutron.db import db_base_plugin_common
 from neutron.db import models_v2
 from neutron.ipam import utils as ipam_utils
@@ -297,9 +298,12 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
                         pool_2=r_range,
                         subnet_cidr=subnet_cidr)
 
-    def _validate_max_ips_per_port(self, fixed_ip_list):
+    def _validate_max_ips_per_port(self, fixed_ip_list, device_owner):
+        if common_utils.is_port_trusted({'device_owner': device_owner}):
+            return
+
         if len(fixed_ip_list) > cfg.CONF.max_fixed_ips_per_port:
-            msg = _('Exceeded maximum amount of fixed ips per port')
+            msg = _('Exceeded maximum amount of fixed ips per port.')
             raise n_exc.InvalidInput(error_message=msg)
 
     def _get_subnet_for_fixed_ip(self, context, fixed, network_id):
@@ -369,7 +373,7 @@ class IpamBackendMixin(db_base_plugin_common.DbBasePluginCommon):
                                   new_ips, device_owner):
         """Calculate changes in IPs for the port."""
         # the new_ips contain all of the fixed_ips that are to be updated
-        self._validate_max_ips_per_port(new_ips)
+        self._validate_max_ips_per_port(new_ips, device_owner)
 
         add_ips = []
         remove_ips = []

neutron/db/ipam_non_pluggable_backend.py

@@ -274,7 +274,7 @@ class IpamNonPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
                     not is_auto_addr_subnet):
                     fixed_ip_set.append({'subnet_id': subnet['id']})
 
-        self._validate_max_ips_per_port(fixed_ip_set)
+        self._validate_max_ips_per_port(fixed_ip_set, device_owner)
         return fixed_ip_set
 
     def _allocate_fixed_ips(self, context, fixed_ips, mac_address):

neutron/db/ipam_pluggable_backend.py

@@ -263,7 +263,7 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
                         not is_auto_addr_subnet):
                     fixed_ip_list.append({'subnet_id': subnet['id']})
 
-        self._validate_max_ips_per_port(fixed_ip_list)
+        self._validate_max_ips_per_port(fixed_ip_list, device_owner)
         return fixed_ip_list
 
     def _update_ips_for_port(self, context, port,

neutron/tests/unit/db/test_db_base_plugin_v2.py

@@ -959,6 +959,23 @@ class TestPortsV2(NeutronDbPluginV2TestCase):
                 self.assertEqual(expected_error, data['NeutronError']['type'])
                 self.assertEqual(msg, data['NeutronError']['message'])
 
+    def test_create_port_with_too_many_fixed_ips(self):
+        with self.network() as network:
+            with self.subnet(network=network, cidr='10.0.0.0/24') as subnet:
+                fixed_ips = [{'subnet_id': subnet['subnet']['id'],
+                              'ip_address': '10.0.0.%s' % id}
+                             for id in range(3,
+                                 cfg.CONF.max_fixed_ips_per_port + 4)]
+                res = self._create_port(self.fmt,
+                                        network['network']['id'],
+                                        webob.exc.HTTPBadRequest.code,
+                                        fixed_ips=fixed_ips,
+                                        set_context=True)
+                data = self.deserialize(self.fmt, res)
+                expected_error = 'InvalidInput'
+                self.assertEqual(expected_error,
+                                 data['NeutronError']['type'])
+
     def test_create_ports_bulk_native(self):
         if self._skip_native_bulk:
             self.skipTest("Plugin does not support native bulk port create")
@@ -1250,6 +1267,32 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
         # sub-classes for plugins/drivers that support mac address update
         # override this method
 
+    def test_update_dhcp_port_with_exceeding_fixed_ips(self):
+        """
+        Max fixed ips per port is configured in configuration file
+        by max_fixed_ips_per_port parameter.
+
+        DHCP port is not restricted by this parameter.
+        """
+        with self.subnet() as subnet:
+            updated_fixed_ips = [{'subnet_id': subnet['subnet']['id'],
+                                  'ip_address': '10.0.0.%s' % id}
+                                 for id in range(3,
+                                     cfg.CONF.max_fixed_ips_per_port + 4)]
+            host_arg = None or {}
+            arg_list = None or []
+            with self.port(device_owner=constants.DEVICE_OWNER_DHCP,
+                           subnet=subnet, arg_list=arg_list,
+                           **host_arg) as port:
+                data = {'port': {'fixed_ips': updated_fixed_ips}}
+                req = self.new_update_request('ports',
+                                              data, port['port']['id'])
+                res = req.get_response(self.api)
+                self.assertEqual(res.status_int, webob.exc.HTTPOk.code)
+                result = self.deserialize(self.fmt, res)
+                for fixed_ip in updated_fixed_ips:
+                    self.assertIn(fixed_ip, result['port']['fixed_ips'])
+
     def test_update_port_mac_ip(self):
         with self.subnet() as subnet:
             updated_fixed_ips = [{'subnet_id': subnet['subnet']['id'],