This commit is a preparation step for using pluggable IPAM.
_update_ips_for_port was refactored and split into two methods:
- _get_changed_ips_for_port
This method contains calculations common for pluggable and
non-pluggable IPAM implementation, was moved to ipam_backend_mixin.
- _update_ips_for_port
This method is specific for non-pluggable IPAM implementation, so it
was moved to ipam_non_pluggable_backend_common.
Other changes:
- _update_ips_for_port now returns namedtuple with added, removed, original
ips (previously added and original ips were returned).
List of removed ips is required by pluggable IPAM implementaion
to apply rollback-on-failure logic;
- removed unused port_id argument from _update_ips_for_port argument list;
Partially-Implements: blueprint neutron-ipam
Change-Id: Id50b6227c8c2d94c35473aece080a6f106a5dfd8
return True
return False
- def _update_ips_for_port(self, context, network_id, port_id, original_ips,
- new_ips, mac_address, device_owner):
- """Add or remove IPs from the port."""
- ips = []
- # These ips are still on the port and haven't been removed
- prev_ips = []
-
- # the new_ips contain all of the fixed_ips that are to be updated
- if len(new_ips) > cfg.CONF.max_fixed_ips_per_port:
- msg = _('Exceeded maximim amount of fixed ips per port')
- raise n_exc.InvalidInput(error_message=msg)
-
- # Remove all of the intersecting elements
- for original_ip in original_ips[:]:
- for new_ip in new_ips[:]:
- if ('ip_address' in new_ip and
- original_ip['ip_address'] == new_ip['ip_address']):
- original_ips.remove(original_ip)
- new_ips.remove(new_ip)
- prev_ips.append(original_ip)
- break
- else:
- # For ports that are not router ports, retain any automatic
- # (non-optional, e.g. IPv6 SLAAC) addresses.
- if device_owner not in constants.ROUTER_INTERFACE_OWNERS:
- subnet = self._get_subnet(context,
- original_ip['subnet_id'])
- if (ipv6_utils.is_auto_address_subnet(subnet)):
- original_ips.remove(original_ip)
- prev_ips.append(original_ip)
-
- # Check if the IP's to add are OK
- to_add = self._test_fixed_ips_for_port(context, network_id, new_ips,
- device_owner)
- for ip in original_ips:
- LOG.debug("Port update. Hold %s", ip)
- NeutronDbPluginV2._delete_ip_allocation(context,
- network_id,
- ip['subnet_id'],
- ip['ip_address'])
-
- if to_add:
- LOG.debug("Port update. Adding %s", to_add)
- ips = self._allocate_fixed_ips(context, to_add, mac_address)
- return ips, prev_ips
-
def _validate_subnet_cidr(self, context, network, new_subnet_cidr):
"""Validate the CIDR for a subnet.
if 'fixed_ips' in p:
changed_ips = True
original = self._make_port_dict(port, process_extensions=False)
- added_ips, prev_ips = self._update_ips_for_port(
- context, network_id, id,
+ changes = self._update_ips_for_port(
+ context, network_id,
original["fixed_ips"], p['fixed_ips'],
original['mac_address'], port['device_owner'])
# Update ips if necessary
- for ip in added_ips:
+ for ip in changes.add:
NeutronDbPluginV2._store_ip_allocation(
context, ip['ip_address'], network_id,
ip['subnet_id'], port.id)
result = self._make_port_dict(port)
# Keep up with fields that changed
if changed_ips:
- result['fixed_ips'] = prev_ips + added_ips
+ result['fixed_ips'] = changes.original + changes.add
return result
def delete_port(self, context, id):
# License for the specific language governing permissions and limitations
# under the License.
-import netaddr
+import collections
+import netaddr
+from oslo_config import cfg
from oslo_log import log as logging
+from neutron.common import constants
from neutron.common import exceptions as n_exc
+from neutron.common import ipv6_utils
from neutron.db import db_base_plugin_common
from neutron.db import models_v2
from neutron.i18n import _LI
"""Contains IPAM specific code which is common for both backends.
"""
+ # Tracks changes in ip allocation for port using namedtuple
+ Changes = collections.namedtuple('Changes', 'add original remove')
+
def _update_subnet_host_routes(self, context, id, s):
def _combine(ht):
raise n_exc.GatewayConflictWithAllocationPools(
pool=pool_range,
ip_address=gateway_ip)
+
+ def _get_changed_ips_for_port(self, context, original_ips,
+ 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
+ if len(new_ips) > cfg.CONF.max_fixed_ips_per_port:
+ msg = _('Exceeded maximum amount of fixed ips per port')
+ raise n_exc.InvalidInput(error_message=msg)
+
+ # These ips are still on the port and haven't been removed
+ prev_ips = []
+
+ # Remove all of the intersecting elements
+ for original_ip in original_ips[:]:
+ for new_ip in new_ips[:]:
+ if ('ip_address' in new_ip and
+ original_ip['ip_address'] == new_ip['ip_address']):
+ original_ips.remove(original_ip)
+ new_ips.remove(new_ip)
+ prev_ips.append(original_ip)
+ break
+ else:
+ # For ports that are not router ports, retain any automatic
+ # (non-optional, e.g. IPv6 SLAAC) addresses.
+ if device_owner not in constants.ROUTER_INTERFACE_OWNERS:
+ subnet = self._get_subnet(context,
+ original_ip['subnet_id'])
+ if (ipv6_utils.is_auto_address_subnet(subnet)):
+ original_ips.remove(original_ip)
+ prev_ips.append(original_ip)
+ return self.Changes(add=new_ips,
+ original=prev_ips,
+ remove=original_ips)
'subnet_id': result['subnet_id']})
return ips
+ def _update_ips_for_port(self, context, network_id, original_ips,
+ new_ips, mac_address, device_owner):
+ """Add or remove IPs from the port."""
+ added = []
+ changes = self._get_changed_ips_for_port(context, original_ips,
+ new_ips, device_owner)
+ # Check if the IP's to add are OK
+ to_add = self._test_fixed_ips_for_port(context, network_id,
+ changes.add, device_owner)
+ for ip in changes.remove:
+ LOG.debug("Port update. Hold %s", ip)
+ IpamNonPluggableBackend._delete_ip_allocation(context,
+ network_id,
+ ip['subnet_id'],
+ ip['ip_address'])
+
+ if to_add:
+ LOG.debug("Port update. Adding %s", to_add)
+ added = self._allocate_fixed_ips(context, to_add, mac_address)
+ return self.Changes(add=added,
+ original=changes.original,
+ remove=changes.remove)
+
def _allocate_ips_for_port(self, context, port):
"""Allocate IP addresses for the port.
--- /dev/null
+# Copyright (c) 2015 Infoblox Inc.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import mock
+
+from neutron.common import constants
+from neutron.db import ipam_backend_mixin
+from neutron.tests import base
+
+
+class TestIpamBackendMixin(base.BaseTestCase):
+
+ def setUp(self):
+ super(TestIpamBackendMixin, self).setUp()
+ self.mixin = ipam_backend_mixin.IpamBackendMixin()
+ self.ctx = mock.Mock()
+ self.default_new_ips = (('id-1', '192.168.1.1'),
+ ('id-2', '192.168.1.2'))
+ self.default_original_ips = (('id-1', '192.168.1.1'),
+ ('id-5', '172.20.16.5'))
+
+ def _prepare_ips(self, ips):
+ return [{'ip_address': ip[1],
+ 'subnet_id': ip[0]} for ip in ips]
+
+ def _test_get_changed_ips_for_port(self, expected_change, original_ips,
+ new_ips, owner):
+ change = self.mixin._get_changed_ips_for_port(self.ctx,
+ original_ips,
+ new_ips,
+ owner)
+ self.assertEqual(expected_change, change)
+
+ def test__get_changed_ips_for_port(self):
+ owner_router = constants.DEVICE_OWNER_ROUTER_INTF
+ new_ips = self._prepare_ips(self.default_new_ips)
+ original_ips = self._prepare_ips(self.default_original_ips)
+
+ # generate changes before calling _get_changed_ips_for_port
+ # because new_ips and original_ips are affected during call
+ expected_change = self.mixin.Changes(add=[new_ips[1]],
+ original=[original_ips[0]],
+ remove=[original_ips[1]])
+ self._test_get_changed_ips_for_port(expected_change, original_ips,
+ new_ips, owner_router)
+
+ def test__get_changed_ips_for_port_autoaddress(self):
+ owner_not_router = constants.DEVICE_OWNER_DHCP
+ new_ips = self._prepare_ips(self.default_new_ips)
+
+ original = (('id-1', '192.168.1.1'),
+ ('id-5', '2000:1234:5678::12FF:FE34:5678'))
+ original_ips = self._prepare_ips(original)
+
+ # mock to test auto address part
+ slaac_subnet = {'ipv6_address_mode': constants.IPV6_SLAAC,
+ 'ipv6_ra_mode': constants.IPV6_SLAAC}
+ self.mixin._get_subnet = mock.Mock(return_value=slaac_subnet)
+
+ # make a copy of original_ips
+ # since it is changed by _get_changed_ips_for_port
+ expected_change = self.mixin.Changes(add=[new_ips[1]],
+ original=original_ips[:],
+ remove=[])
+
+ self._test_get_changed_ips_for_port(expected_change, original_ips,
+ new_ips, owner_not_router)
Code Review / openstack-build / neutron-build.git / commitdiff