Introduces new Pluggable IPAM backend.
IP/subnet allocation calls are sent to IPAM driver.
Calls to IPAM Driver are considered as call to third-party
environment, so if any action fails, rollback action is called.
Removes associate_neutron_subnet step from interface and reference
driver. It is not needed any more because foreign key relationship
between IPAM subnet and neutron subnet was removed.
So IPAM subnet can store id of neutron subnet,
which is not created yet.
For now only reference IPAM driver is available.
Temporarily disabled test_ipam_pluggable_backend from
gate-neutron-python34. Current patch adds executing
parts of test_db_base_plugin_v2, which is not py34 compatible yet.
Might be enabled back once 204791 is merged.
Partially-Implements: blueprint neutron-ipam
Change-Id: Ic18461cf19d2eebf8fad5535eee0eb034959800e
# DEVICE_OWNER_ROUTER_HA_INTF is a special case and so is not included.
ROUTER_INTERFACE_OWNERS = (DEVICE_OWNER_ROUTER_INTF,
DEVICE_OWNER_DVR_INTERFACE)
+ROUTER_INTERFACE_OWNERS_SNAT = (DEVICE_OWNER_ROUTER_INTF,
+ DEVICE_OWNER_DVR_INTERFACE,
+ DEVICE_OWNER_ROUTER_SNAT)
L3_AGENT_MODE_DVR = 'dvr'
L3_AGENT_MODE_DVR_SNAT = 'dvr_snat'
L3_AGENT_MODE_LEGACY = 'legacy'
from neutron.db import api as db_api
from neutron.db import db_base_plugin_common
from neutron.db import ipam_non_pluggable_backend
+from neutron.db import ipam_pluggable_backend
from neutron.db import models_v2
from neutron.db import rbac_db_models as rbac_db
from neutron.db import sqlalchemyutils
self.nova_notifier.record_port_status_changed)
def set_ipam_backend(self):
- self.ipam = ipam_non_pluggable_backend.IpamNonPluggableBackend()
+ if cfg.CONF.ipam_driver:
+ self.ipam = ipam_pluggable_backend.IpamPluggableBackend()
+ else:
+ self.ipam = ipam_non_pluggable_backend.IpamNonPluggableBackend()
def _validate_host_route(self, route, ip_version):
try:
with context.session.begin(subtransactions=True):
network = self._get_network(context, s["network_id"])
- subnet = self.ipam.allocate_subnet(context,
- network,
- s,
- subnetpool_id)
+ subnet, ipam_subnet = self.ipam.allocate_subnet(context,
+ network,
+ s,
+ subnetpool_id)
if hasattr(network, 'external') and network.external:
self._update_router_gw_ports(context,
network,
# If this subnet supports auto-addressing, then update any
# internal ports on the network with addresses for this subnet.
if ipv6_utils.is_auto_address_subnet(subnet):
- self.ipam.add_auto_addrs_on_network_ports(context, subnet)
+ self.ipam.add_auto_addrs_on_network_ports(context, subnet,
+ ipam_subnet)
return self._make_subnet_dict(subnet, context=context)
def _get_subnetpool_id(self, subnet):
s['ip_version'] = db_subnet.ip_version
s['cidr'] = db_subnet.cidr
s['id'] = db_subnet.id
+ s['tenant_id'] = db_subnet.tenant_id
self._validate_subnet(context, s, cur_subnet=db_subnet)
+ db_pools = [netaddr.IPRange(p['first_ip'], p['last_ip'])
+ for p in db_subnet.allocation_pools]
+
+ range_pools = None
+ if s.get('allocation_pools') is not None:
+ # Convert allocation pools to IPRange to simplify future checks
+ range_pools = self.ipam.pools_to_ip_range(s['allocation_pools'])
+ s['allocation_pools'] = range_pools
if s.get('gateway_ip') is not None:
- if s.get('allocation_pools') is not None:
- allocation_pools = [{'start': p['start'], 'end': p['end']}
- for p in s['allocation_pools']]
- else:
- allocation_pools = [{'start': p['first_ip'],
- 'end': p['last_ip']}
- for p in db_subnet.allocation_pools]
- self.ipam.validate_gw_out_of_pools(s["gateway_ip"],
- allocation_pools)
+ pools = range_pools if range_pools is not None else db_pools
+ self.ipam.validate_gw_out_of_pools(s["gateway_ip"], pools)
with context.session.begin(subtransactions=True):
- subnet, changes = self.ipam.update_db_subnet(context, id, s)
+ subnet, changes = self.ipam.update_db_subnet(context, id, s,
+ db_pools)
result = self._make_subnet_dict(subnet, context=context)
# Keep up with fields that changed
result.update(changes)
raise n_exc.SubnetInUse(subnet_id=id)
context.session.delete(subnet)
+ # Delete related ipam subnet manually,
+ # since there is no FK relationship
+ self.ipam.delete_subnet(context, id)
def get_subnet(self, context, id, fields=None):
subnet = self._get_subnet(context, id)
return str(netaddr.IPNetwork(cidr_net).network + 1)
return subnet.get('gateway_ip')
+ @staticmethod
+ def pools_to_ip_range(ip_pools):
+ ip_range_pools = []
+ for ip_pool in ip_pools:
+ try:
+ ip_range_pools.append(netaddr.IPRange(ip_pool['start'],
+ ip_pool['end']))
+ except netaddr.AddrFormatError:
+ LOG.info(_LI("Found invalid IP address in pool: "
+ "%(start)s - %(end)s:"),
+ {'start': ip_pool['start'],
+ 'end': ip_pool['end']})
+ raise n_exc.InvalidAllocationPool(pool=ip_pool)
+ return ip_range_pools
+
+ def delete_subnet(self, context, subnet_id):
+ pass
+
def validate_pools_with_subnetpool(self, subnet):
"""Verifies that allocation pools are set correctly
def _update_subnet_allocation_pools(self, context, subnet_id, s):
context.session.query(models_v2.IPAllocationPool).filter_by(
subnet_id=subnet_id).delete()
- new_pools = [models_v2.IPAllocationPool(first_ip=p['start'],
- last_ip=p['end'],
+ pools = ((netaddr.IPAddress(p.first, p.version).format(),
+ netaddr.IPAddress(p.last, p.version).format())
+ for p in s['allocation_pools'])
+ new_pools = [models_v2.IPAllocationPool(first_ip=p[0],
+ last_ip=p[1],
subnet_id=subnet_id)
- for p in s['allocation_pools']]
+ for p in pools]
context.session.add_all(new_pools)
# Call static method with self to redefine in child
# (non-pluggable backend)
self._rebuild_availability_ranges(context, [s])
- # Gather new pools for result:
- result_pools = [{'start': pool['start'],
- 'end': pool['end']}
- for pool in s['allocation_pools']]
+ # Gather new pools for result
+ result_pools = [{'start': p[0], 'end': p[1]} for p in pools]
del s['allocation_pools']
return result_pools
- def update_db_subnet(self, context, subnet_id, s):
+ def update_db_subnet(self, context, subnet_id, s, oldpools):
changes = {}
if "dns_nameservers" in s:
changes['dns_nameservers'] = (
LOG.debug("Performing IP validity checks on allocation pools")
ip_sets = []
for ip_pool in ip_pools:
- try:
- start_ip = netaddr.IPAddress(ip_pool['start'])
- end_ip = netaddr.IPAddress(ip_pool['end'])
- except netaddr.AddrFormatError:
- LOG.info(_LI("Found invalid IP address in pool: "
- "%(start)s - %(end)s:"),
- {'start': ip_pool['start'],
- 'end': ip_pool['end']})
- raise n_exc.InvalidAllocationPool(pool=ip_pool)
+ start_ip = netaddr.IPAddress(ip_pool.first, ip_pool.version)
+ end_ip = netaddr.IPAddress(ip_pool.last, ip_pool.version)
if (start_ip.version != subnet.version or
end_ip.version != subnet.version):
LOG.info(_LI("Specified IP addresses do not match "
"the subnet IP version"))
raise n_exc.InvalidAllocationPool(pool=ip_pool)
- if end_ip < start_ip:
- LOG.info(_LI("Start IP (%(start)s) is greater than end IP "
- "(%(end)s)"),
- {'start': ip_pool['start'], 'end': ip_pool['end']})
- raise n_exc.InvalidAllocationPool(pool=ip_pool)
if start_ip < subnet_first_ip or end_ip > subnet_last_ip:
LOG.info(_LI("Found pool larger than subnet "
"CIDR:%(start)s - %(end)s"),
- {'start': ip_pool['start'],
- 'end': ip_pool['end']})
+ {'start': start_ip, 'end': end_ip})
raise n_exc.OutOfBoundsAllocationPool(
pool=ip_pool,
subnet_cidr=subnet_cidr)
# Valid allocation pool
# Create an IPSet for it for easily verifying overlaps
- ip_sets.append(netaddr.IPSet(netaddr.IPRange(
- ip_pool['start'],
- ip_pool['end']).cidrs()))
+ ip_sets.append(netaddr.IPSet(ip_pool.cidrs()))
LOG.debug("Checking for overlaps among allocation pools "
"and gateway ip")
pool_2=r_range,
subnet_cidr=subnet_cidr)
+ def _validate_max_ips_per_port(self, fixed_ip_list):
+ if len(fixed_ip_list) > cfg.CONF.max_fixed_ips_per_port:
+ msg = _('Exceeded maximim amount of fixed ips per port')
+ raise n_exc.InvalidInput(error_message=msg)
+
+ def _get_subnet_for_fixed_ip(self, context, fixed, network_id):
+ if 'subnet_id' in fixed:
+ subnet = self._get_subnet(context, fixed['subnet_id'])
+ if subnet['network_id'] != network_id:
+ msg = (_("Failed to create port on network %(network_id)s"
+ ", because fixed_ips included invalid subnet "
+ "%(subnet_id)s") %
+ {'network_id': network_id,
+ 'subnet_id': fixed['subnet_id']})
+ raise n_exc.InvalidInput(error_message=msg)
+ # Ensure that the IP is valid on the subnet
+ if ('ip_address' in fixed and
+ not ipam_utils.check_subnet_ip(subnet['cidr'],
+ fixed['ip_address'])):
+ raise n_exc.InvalidIpForSubnet(ip_address=fixed['ip_address'])
+ return subnet
+
+ if 'ip_address' not in fixed:
+ msg = _('IP allocation requires subnet_id or ip_address')
+ raise n_exc.InvalidInput(error_message=msg)
+
+ filter = {'network_id': [network_id]}
+ subnets = self._get_subnets(context, filters=filter)
+
+ for subnet in subnets:
+ if ipam_utils.check_subnet_ip(subnet['cidr'],
+ fixed['ip_address']):
+ return subnet
+ raise n_exc.InvalidIpForNetwork(ip_address=fixed['ip_address'])
+
def _prepare_allocation_pools(self, allocation_pools, cidr, gateway_ip):
"""Returns allocation pools represented as list of IPRanges"""
if not attributes.is_attr_set(allocation_pools):
return ipam_utils.generate_pools(cidr, gateway_ip)
- self._validate_allocation_pools(allocation_pools, cidr)
+ ip_range_pools = self.pools_to_ip_range(allocation_pools)
+ self._validate_allocation_pools(ip_range_pools, cidr)
if gateway_ip:
- self.validate_gw_out_of_pools(gateway_ip, allocation_pools)
- return [netaddr.IPRange(p['start'], p['end'])
- for p in allocation_pools]
+ self.validate_gw_out_of_pools(gateway_ip, ip_range_pools)
+ return ip_range_pools
def validate_gw_out_of_pools(self, gateway_ip, pools):
- for allocation_pool in pools:
- pool_range = netaddr.IPRange(
- allocation_pool['start'],
- allocation_pool['end'])
+ for pool_range in pools:
if netaddr.IPAddress(gateway_ip) in pool_range:
raise n_exc.GatewayConflictWithAllocationPools(
pool=pool_range,
# under the License.
import netaddr
-from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
from sqlalchemy import and_
from neutron.db import models_v2
from neutron.ipam import requests as ipam_req
from neutron.ipam import subnet_alloc
-from neutron.ipam import utils as ipam_utils
LOG = logging.getLogger(__name__)
"""
fixed_ip_set = []
for fixed in fixed_ips:
- found = False
- if 'subnet_id' not in fixed:
- if 'ip_address' not in fixed:
- msg = _('IP allocation requires subnet_id or ip_address')
- raise n_exc.InvalidInput(error_message=msg)
-
- filter = {'network_id': [network_id]}
- subnets = self._get_subnets(context, filters=filter)
- for subnet in subnets:
- if ipam_utils.check_subnet_ip(subnet['cidr'],
- fixed['ip_address']):
- found = True
- subnet_id = subnet['id']
- break
- if not found:
- raise n_exc.InvalidIpForNetwork(
- ip_address=fixed['ip_address'])
- else:
- subnet = self._get_subnet(context, fixed['subnet_id'])
- if subnet['network_id'] != network_id:
- msg = (_("Failed to create port on network %(network_id)s"
- ", because fixed_ips included invalid subnet "
- "%(subnet_id)s") %
- {'network_id': network_id,
- 'subnet_id': fixed['subnet_id']})
- raise n_exc.InvalidInput(error_message=msg)
- subnet_id = subnet['id']
+ subnet = self._get_subnet_for_fixed_ip(context, fixed, network_id)
is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
if 'ip_address' in fixed:
# Ensure that the IP's are unique
if not IpamNonPluggableBackend._check_unique_ip(
context, network_id,
- subnet_id, fixed['ip_address']):
+ subnet['id'], fixed['ip_address']):
raise n_exc.IpAddressInUse(net_id=network_id,
ip_address=fixed['ip_address'])
- # Ensure that the IP is valid on the subnet
- if (not found and
- not ipam_utils.check_subnet_ip(subnet['cidr'],
- fixed['ip_address'])):
- raise n_exc.InvalidIpForSubnet(
- ip_address=fixed['ip_address'])
if (is_auto_addr_subnet and
device_owner not in
constants.ROUTER_INTERFACE_OWNERS):
"assigned to a port on subnet %(id)s since the "
"subnet is configured for automatic addresses") %
{'address': fixed['ip_address'],
- 'id': subnet_id})
+ 'id': subnet['id']})
raise n_exc.InvalidInput(error_message=msg)
- fixed_ip_set.append({'subnet_id': subnet_id,
+ fixed_ip_set.append({'subnet_id': subnet['id'],
'ip_address': fixed['ip_address']})
else:
# A scan for auto-address subnets on the network is done
# separately so that all such subnets (not just those
# listed explicitly here by subnet ID) are associated
# with the port.
- if (device_owner in constants.ROUTER_INTERFACE_OWNERS or
- device_owner == constants.DEVICE_OWNER_ROUTER_SNAT or
+ if (device_owner in constants.ROUTER_INTERFACE_OWNERS_SNAT or
not is_auto_addr_subnet):
- fixed_ip_set.append({'subnet_id': subnet_id})
+ fixed_ip_set.append({'subnet_id': subnet['id']})
- if len(fixed_ip_set) > cfg.CONF.max_fixed_ips_per_port:
- msg = _('Exceeded maximim amount of fixed ips per port')
- raise n_exc.InvalidInput(error_message=msg)
+ self._validate_max_ips_per_port(fixed_ip_set)
return fixed_ip_set
def _allocate_fixed_ips(self, context, fixed_ips, mac_address):
net_id_filter = {'network_id': [p['network_id']]}
subnets = self._get_subnets(context, filters=net_id_filter)
is_router_port = (
- p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS or
- p['device_owner'] == constants.DEVICE_OWNER_ROUTER_SNAT)
+ p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
fixed_configured = p['fixed_ips'] is not attributes.ATTR_NOT_SPECIFIED
if fixed_configured:
return ips
- def add_auto_addrs_on_network_ports(self, context, subnet):
+ def add_auto_addrs_on_network_ports(self, context, subnet, ipam_subnet):
"""For an auto-address subnet, add addrs for ports on the net."""
with context.session.begin(subtransactions=True):
network_id = subnet['network_id']
port_qry = context.session.query(models_v2.Port)
- for port in port_qry.filter(
+ ports = port_qry.filter(
and_(models_v2.Port.network_id == network_id,
- models_v2.Port.device_owner !=
- constants.DEVICE_OWNER_ROUTER_SNAT,
~models_v2.Port.device_owner.in_(
- constants.ROUTER_INTERFACE_OWNERS))):
+ constants.ROUTER_INTERFACE_OWNERS_SNAT)))
+ for port in ports:
ip_address = self._calculate_ipv6_eui64_addr(
context, subnet, port['mac_address'])
allocated = models_v2.IPAllocation(network_id=network_id,
subnet['dns_nameservers'],
subnet['host_routes'],
subnet_request)
- return subnet
+ # ipam_subnet is not expected to be allocated for non pluggable ipam,
+ # so just return None for it (second element in returned tuple)
+ return subnet, None
# License for the specific language governing permissions and limitations
# under the License.
+import netaddr
+from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_utils import excutils
+from sqlalchemy import and_
+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.db import ipam_backend_mixin
+from neutron.db import models_v2
from neutron.i18n import _LE
+from neutron.ipam import driver
from neutron.ipam import exceptions as ipam_exc
+from neutron.ipam import requests as ipam_req
LOG = logging.getLogger(__name__)
ip_address, ip_subnet = self._ipam_allocate_single_ip(
context, ipam_driver, port, ip_list)
allocated.append({'ip_address': ip_address,
- 'subnet_cidr': ip_subnet['subnet_cidr'],
'subnet_id': ip_subnet['subnet_id']})
except Exception:
with excutils.save_and_reraise_exception():
"external system for %s"), addresses)
return allocated
+
+ def _ipam_update_allocation_pools(self, context, ipam_driver, subnet):
+ self._validate_allocation_pools(subnet['allocation_pools'],
+ subnet['cidr'])
+
+ factory = ipam_driver.get_subnet_request_factory()
+ subnet_request = factory.get_request(context, subnet, None)
+
+ ipam_driver.update_subnet(subnet_request)
+
+ def delete_subnet(self, context, subnet_id):
+ ipam_driver = driver.Pool.get_instance(None, context)
+ ipam_driver.remove_subnet(subnet_id)
+
+ def allocate_ips_for_port_and_store(self, context, port, port_id):
+ network_id = port['port']['network_id']
+ ips = []
+ try:
+ ips = self._allocate_ips_for_port(context, port)
+ for ip in ips:
+ ip_address = ip['ip_address']
+ subnet_id = ip['subnet_id']
+ IpamPluggableBackend._store_ip_allocation(
+ context, ip_address, network_id,
+ subnet_id, port_id)
+ except Exception:
+ with excutils.save_and_reraise_exception():
+ if ips:
+ LOG.debug("An exception occurred during port creation."
+ "Reverting IP allocation")
+ ipam_driver = driver.Pool.get_instance(None, context)
+ self._ipam_deallocate_ips(context, ipam_driver,
+ port['port'], ips,
+ revert_on_fail=False)
+
+ def _allocate_ips_for_port(self, context, port):
+ """Allocate IP addresses for the port. IPAM version.
+
+ If port['fixed_ips'] is set to 'ATTR_NOT_SPECIFIED', allocate IP
+ addresses for the port. If port['fixed_ips'] contains an IP address or
+ a subnet_id then allocate an IP address accordingly.
+ """
+ p = port['port']
+ ips = []
+ v6_stateless = []
+ net_id_filter = {'network_id': [p['network_id']]}
+ subnets = self._get_subnets(context, filters=net_id_filter)
+ is_router_port = (
+ p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
+
+ fixed_configured = p['fixed_ips'] is not attributes.ATTR_NOT_SPECIFIED
+ if fixed_configured:
+ ips = self._test_fixed_ips_for_port(context,
+ p["network_id"],
+ p['fixed_ips'],
+ p['device_owner'])
+ # For ports that are not router ports, implicitly include all
+ # auto-address subnets for address association.
+ if not is_router_port:
+ v6_stateless += [subnet for subnet in subnets
+ if ipv6_utils.is_auto_address_subnet(subnet)]
+ else:
+ # Split into v4, v6 stateless and v6 stateful subnets
+ v4 = []
+ v6_stateful = []
+ for subnet in subnets:
+ if subnet['ip_version'] == 4:
+ v4.append(subnet)
+ else:
+ if ipv6_utils.is_auto_address_subnet(subnet):
+ if not is_router_port:
+ v6_stateless.append(subnet)
+ else:
+ v6_stateful.append(subnet)
+
+ version_subnets = [v4, v6_stateful]
+ for subnets in version_subnets:
+ if subnets:
+ ips.append([{'subnet_id': s['id']}
+ for s in subnets])
+
+ for subnet in v6_stateless:
+ # IP addresses for IPv6 SLAAC and DHCPv6-stateless subnets
+ # are implicitly included.
+ ips.append({'subnet_id': subnet['id'],
+ 'subnet_cidr': subnet['cidr'],
+ 'eui64_address': True,
+ 'mac': p['mac_address']})
+ ipam_driver = driver.Pool.get_instance(None, context)
+ return self._ipam_allocate_ips(context, ipam_driver, p, ips)
+
+ def _test_fixed_ips_for_port(self, context, network_id, fixed_ips,
+ device_owner):
+ """Test fixed IPs for port.
+
+ Check that configured subnets are valid prior to allocating any
+ IPs. Include the subnet_id in the result if only an IP address is
+ configured.
+
+ :raises: InvalidInput, IpAddressInUse, InvalidIpForNetwork,
+ InvalidIpForSubnet
+ """
+ fixed_ip_list = []
+ for fixed in fixed_ips:
+ subnet = self._get_subnet_for_fixed_ip(context, fixed, network_id)
+
+ is_auto_addr_subnet = ipv6_utils.is_auto_address_subnet(subnet)
+ if 'ip_address' in fixed:
+ if (is_auto_addr_subnet and device_owner not in
+ constants.ROUTER_INTERFACE_OWNERS):
+ msg = (_("IPv6 address %(address)s can not be directly "
+ "assigned to a port on subnet %(id)s since the "
+ "subnet is configured for automatic addresses") %
+ {'address': fixed['ip_address'],
+ 'id': subnet['id']})
+ raise n_exc.InvalidInput(error_message=msg)
+ fixed_ip_list.append({'subnet_id': subnet['id'],
+ 'ip_address': fixed['ip_address']})
+ else:
+ # A scan for auto-address subnets on the network is done
+ # separately so that all such subnets (not just those
+ # listed explicitly here by subnet ID) are associated
+ # with the port.
+ if (device_owner in constants.ROUTER_INTERFACE_OWNERS_SNAT or
+ not is_auto_addr_subnet):
+ fixed_ip_list.append({'subnet_id': subnet['id']})
+
+ self._validate_max_ips_per_port(fixed_ip_list)
+ return fixed_ip_list
+
+ def _update_ips_for_port(self, context, port,
+ original_ips, new_ips, mac):
+ """Add or remove IPs from the port. IPAM version"""
+ added = []
+ removed = []
+ changes = self._get_changed_ips_for_port(
+ context, original_ips, new_ips, port['device_owner'])
+ # Check if the IP's to add are OK
+ to_add = self._test_fixed_ips_for_port(
+ context, port['network_id'], changes.add,
+ port['device_owner'])
+
+ ipam_driver = driver.Pool.get_instance(None, context)
+ if changes.remove:
+ removed = self._ipam_deallocate_ips(context, ipam_driver, port,
+ changes.remove)
+ if to_add:
+ added = self._ipam_allocate_ips(context, ipam_driver,
+ changes, to_add)
+ return self.Changes(add=added,
+ original=changes.original,
+ remove=removed)
+
+ def save_allocation_pools(self, context, subnet, allocation_pools):
+ for pool in allocation_pools:
+ first_ip = str(netaddr.IPAddress(pool.first, pool.version))
+ last_ip = str(netaddr.IPAddress(pool.last, pool.version))
+ ip_pool = models_v2.IPAllocationPool(subnet=subnet,
+ first_ip=first_ip,
+ last_ip=last_ip)
+ context.session.add(ip_pool)
+
+ def update_port_with_ips(self, context, db_port, new_port, new_mac):
+ changes = self.Changes(add=[], original=[], remove=[])
+
+ if 'fixed_ips' in new_port:
+ original = self._make_port_dict(db_port,
+ process_extensions=False)
+ changes = self._update_ips_for_port(context,
+ db_port,
+ original["fixed_ips"],
+ new_port['fixed_ips'],
+ new_mac)
+ try:
+ # Check if the IPs need to be updated
+ network_id = db_port['network_id']
+ for ip in changes.add:
+ self._store_ip_allocation(
+ context, ip['ip_address'], network_id,
+ ip['subnet_id'], db_port.id)
+ for ip in changes.remove:
+ self._delete_ip_allocation(context, network_id,
+ ip['subnet_id'], ip['ip_address'])
+ self._update_db_port(context, db_port, new_port, network_id,
+ new_mac)
+ except Exception:
+ with excutils.save_and_reraise_exception():
+ if 'fixed_ips' in new_port:
+ LOG.debug("An exception occurred during port update.")
+ ipam_driver = driver.Pool.get_instance(None, context)
+ if changes.add:
+ LOG.debug("Reverting IP allocation.")
+ self._ipam_deallocate_ips(context, ipam_driver,
+ db_port, changes.add,
+ revert_on_fail=False)
+ if changes.remove:
+ LOG.debug("Reverting IP deallocation.")
+ self._ipam_allocate_ips(context, ipam_driver,
+ db_port, changes.remove,
+ revert_on_fail=False)
+ return changes
+
+ def delete_port(self, context, id):
+ # Get fixed_ips list before port deletion
+ port = self._get_port(context, id)
+ ipam_driver = driver.Pool.get_instance(None, context)
+
+ super(IpamPluggableBackend, self).delete_port(context, id)
+ # Deallocating ips via IPAM after port is deleted locally.
+ # So no need to do rollback actions on remote server
+ # in case of fail to delete port locally
+ self._ipam_deallocate_ips(context, ipam_driver, port,
+ port['fixed_ips'])
+
+ def update_db_subnet(self, context, id, s, old_pools):
+ ipam_driver = driver.Pool.get_instance(None, context)
+ if "allocation_pools" in s:
+ self._ipam_update_allocation_pools(context, ipam_driver, s)
+
+ try:
+ subnet, changes = super(IpamPluggableBackend,
+ self).update_db_subnet(context, id,
+ s, old_pools)
+ except Exception:
+ with excutils.save_and_reraise_exception():
+ if "allocation_pools" in s and old_pools:
+ LOG.error(
+ _LE("An exception occurred during subnet update."
+ "Reverting allocation pool changes"))
+ s['allocation_pools'] = old_pools
+ self._ipam_update_allocation_pools(context, ipam_driver, s)
+ return subnet, changes
+
+ def add_auto_addrs_on_network_ports(self, context, subnet, ipam_subnet):
+ """For an auto-address subnet, add addrs for ports on the net."""
+ with context.session.begin(subtransactions=True):
+ network_id = subnet['network_id']
+ port_qry = context.session.query(models_v2.Port)
+ ports = port_qry.filter(
+ and_(models_v2.Port.network_id == network_id,
+ ~models_v2.Port.device_owner.in_(
+ constants.ROUTER_INTERFACE_OWNERS_SNAT)))
+ for port in ports:
+ ip_request = ipam_req.AutomaticAddressRequest(
+ prefix=subnet['cidr'],
+ mac=port['mac_address'])
+ ip_address = ipam_subnet.allocate(ip_request)
+ allocated = models_v2.IPAllocation(network_id=network_id,
+ port_id=port['id'],
+ ip_address=ip_address,
+ subnet_id=subnet['id'])
+ try:
+ # Do the insertion of each IP allocation entry within
+ # the context of a nested transaction, so that the entry
+ # is rolled back independently of other entries whenever
+ # the corresponding port has been deleted.
+ with context.session.begin_nested():
+ context.session.add(allocated)
+ except db_exc.DBReferenceError:
+ LOG.debug("Port %s was deleted while updating it with an "
+ "IPv6 auto-address. Ignoring.", port['id'])
+ LOG.debug("Reverting IP allocation for %s", ip_address)
+ # Do not fail if reverting allocation was unsuccessful
+ try:
+ ipam_subnet.deallocate(ip_address)
+ except Exception:
+ LOG.debug("Reverting IP allocation failed for %s",
+ ip_address)
+
+ def allocate_subnet(self, context, network, subnet, subnetpool_id):
+ subnetpool = None
+
+ if subnetpool_id:
+ subnetpool = self._get_subnetpool(context, subnetpool_id)
+ self._validate_ip_version_with_subnetpool(subnet, subnetpool)
+
+ # gateway_ip and allocation pools should be validated or generated
+ # only for specific request
+ if subnet['cidr'] is not attributes.ATTR_NOT_SPECIFIED:
+ subnet['gateway_ip'] = self._gateway_ip_str(subnet,
+ subnet['cidr'])
+ subnet['allocation_pools'] = self._prepare_allocation_pools(
+ subnet['allocation_pools'],
+ subnet['cidr'],
+ subnet['gateway_ip'])
+
+ ipam_driver = driver.Pool.get_instance(subnetpool, context)
+ subnet_factory = ipam_driver.get_subnet_request_factory()
+ subnet_request = subnet_factory.get_request(context, subnet,
+ subnetpool)
+ ipam_subnet = ipam_driver.allocate_subnet(subnet_request)
+ # get updated details with actually allocated subnet
+ subnet_request = ipam_subnet.get_details()
+
+ try:
+ subnet = self._save_subnet(context,
+ network,
+ self._make_subnet_args(
+ subnet_request,
+ subnet,
+ subnetpool_id),
+ subnet['dns_nameservers'],
+ subnet['host_routes'],
+ subnet_request)
+ except Exception:
+ # Note(pbondar): Third-party ipam servers can't rely
+ # on transaction rollback, so explicit rollback call needed.
+ # IPAM part rolled back in exception handling
+ # and subnet part is rolled back by transaction rollback.
+ with excutils.save_and_reraise_exception():
+ LOG.debug("An exception occurred during subnet creation."
+ "Reverting subnet allocation.")
+ self.delete_subnet(context, subnet_request.subnet_id)
+ return subnet, ipam_subnet
:returns: An instance of SpecificSubnetRequest with the subnet detail.
"""
-
- @abc.abstractmethod
- def associate_neutron_subnet(self, subnet_id):
- """Associate the IPAM subnet with a neutron subnet.
-
- This operation should be performed to attach a neutron subnet to the
- current subnet instance. In some cases IPAM subnets may be created
- independently of neutron subnets and associated at a later stage.
-
- :param subnet_id: neutron subnet identifier.
- """
session.add(ipam_subnet)
return self._ipam_subnet_id
- def associate_neutron_id(self, session, neutron_subnet_id):
- session.query(db_models.IpamSubnet).filter_by(
- id=self._ipam_subnet_id).update(
- {'neutron_subnet_id': neutron_subnet_id})
- self._neutron_subnet_id = neutron_subnet_id
+ @classmethod
+ def delete(cls, session, neutron_subnet_id):
+ """Delete IPAM subnet.
+
+ IPAM subnet no longer has foreign key to neutron subnet,
+ so need to perform delete manually
+
+ :param session: database sesssion
+ :param neutron_subnet_id: neutron subnet id associated with ipam subnet
+ """
+ return session.query(db_models.IpamSubnet).filter_by(
+ neutron_subnet_id=neutron_subnet_id).delete()
def create_pool(self, session, pool_start, pool_end):
"""Create an allocation pool and availability ranges for the subnet.
ipam_subnet_id = uuidutils.generate_uuid()
subnet_manager = ipam_db_api.IpamSubnetManager(
ipam_subnet_id,
- None)
+ subnet_request.subnet_id)
# Create subnet resource
session = ctx.session
subnet_manager.create(session)
allocation_pools=pools,
gateway_ip=subnet_request.gateway_ip,
tenant_id=subnet_request.tenant_id,
- subnet_id=subnet_request.subnet_id,
- subnet_id_not_set=True)
+ subnet_id=subnet_request.subnet_id)
@classmethod
def load(cls, neutron_subnet_id, ctx):
ipam_subnet = ipam_db_api.IpamSubnetManager.load_by_neutron_subnet_id(
ctx.session, neutron_subnet_id)
if not ipam_subnet:
- LOG.error(_LE("Unable to retrieve IPAM subnet as the referenced "
+ LOG.error(_LE("IPAM subnet referenced to "
"Neutron subnet %s does not exist"),
neutron_subnet_id)
raise n_exc.SubnetNotFound(subnet_id=neutron_subnet_id)
def __init__(self, internal_id, ctx, cidr=None,
allocation_pools=None, gateway_ip=None, tenant_id=None,
- subnet_id=None, subnet_id_not_set=False):
+ subnet_id=None):
# NOTE: In theory it could have been possible to grant the IPAM
# driver direct access to the database. While this is possible,
# it would have led to duplicate code and/or non-trivial
self._pools = allocation_pools
self._gateway_ip = gateway_ip
self._tenant_id = tenant_id
- self._subnet_id = None if subnet_id_not_set else subnet_id
+ self._subnet_id = subnet_id
self.subnet_manager = ipam_db_api.IpamSubnetManager(internal_id,
self._subnet_id)
self._context = ctx
self._tenant_id, self.subnet_manager.neutron_id,
self._cidr, self._gateway_ip, self._pools)
- def associate_neutron_subnet(self, subnet_id):
- """Set neutron identifier for this subnet"""
- session = self._context.session
- if self._subnet_id:
- raise
- # IPAMSubnet does not have foreign key to Subnet,
- # so need verify subnet existence.
- NeutronDbSubnet._fetch_subnet(self._context, subnet_id)
- self.subnet_manager.associate_neutron_id(session, subnet_id)
- self._subnet_id = subnet_id
-
class NeutronDbPool(subnet_alloc.SubnetAllocator):
"""Subnet pools backed by Neutron Database.
subnet.update_allocation_pools(subnet_request.allocation_pools)
return subnet
- def remove_subnet(self, subnet):
+ def remove_subnet(self, subnet_id):
"""Remove data structures for a given subnet.
- All the IPAM-related data are cleared when a subnet is deleted thanks
- to cascaded foreign key relationships.
+ IPAM-related data has no foreign key relationships to neutron subnet,
+ so removing ipam subnet manually
"""
- pass
+ count = ipam_db_api.IpamSubnetManager.delete(self._context.session,
+ subnet_id)
+ if count < 1:
+ LOG.error(_LE("IPAM subnet referenced to "
+ "Neutron subnet %s does not exist"),
+ subnet_id)
+ raise n_exc.SubnetNotFound(subnet_id=subnet_id)
def get_details(self):
return self._req
- def associate_neutron_subnet(self, subnet_id):
- pass
-
class SubnetPoolReader(object):
'''Class to assist with reading a subnetpool, loading defaults, and
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import model_base
from neutron.db import models_v2
+from neutron.ipam.drivers.neutrondb_ipam import db_models as ipam_models
from neutron.tests import base
from neutron.tests.common import base as common_base
Base class for tests that aim to test ip allocation.
"""
- def configure_test(self):
+ def configure_test(self, use_pluggable_ipam=False):
model_base.BASEV2.metadata.create_all(self.engine)
cfg.CONF.set_override('notify_nova_on_port_status_changes', False)
+ if use_pluggable_ipam:
+ self._turn_on_pluggable_ipam()
+ else:
+ self._turn_off_pluggable_ipam()
self.plugin = base_plugin.NeutronDbPluginV2()
self.cxt = get_admin_test_context(self.engine.url)
self.addCleanup(self.cxt._session.close)
self._create_network()
self._create_subnet()
+ def _turn_off_pluggable_ipam(self):
+ cfg.CONF.set_override('ipam_driver', None)
+ self.ip_availability_range = models_v2.IPAvailabilityRange
+
+ def _turn_on_pluggable_ipam(self):
+ cfg.CONF.set_override('ipam_driver', 'internal')
+ DB_PLUGIN_KLASS = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
+ self.setup_coreplugin(DB_PLUGIN_KLASS)
+ self.ip_availability_range = ipam_models.IpamAvailabilityRange
+
def result_set_to_dicts(self, resultset, keys):
dicts = []
for item in resultset:
def assert_ip_avail_range_matches(self, expected):
result_set = self.cxt.session.query(
- models_v2.IPAvailabilityRange).all()
+ self.ip_availability_range).all()
keys = ['first_ip', 'last_ip']
actual = self.result_set_to_dicts(result_set, keys)
self.assertEqual(expected, actual)
def setUp(self):
super(TestIpamPsql, self).setUp()
self.configure_test()
+
+
+class TestPluggableIpamMySql(common_base.MySQLTestCase,
+ base.BaseTestCase, IpamTestCase):
+
+ def setUp(self):
+ super(TestPluggableIpamMySql, self).setUp()
+ self.configure_test(use_pluggable_ipam=True)
+
+
+class TestPluggableIpamPsql(common_base.PostgreSQLTestCase,
+ base.BaseTestCase, IpamTestCase):
+
+ def setUp(self):
+ super(TestPluggableIpamPsql, self).setUp()
+ self.configure_test(use_pluggable_ipam=True)
[{'start': '10.0.0.2', 'end': '10.0.0.254'},
{'end': '10.0.0.254'}],
None,
+ [{'start': '10.0.0.200', 'end': '10.0.3.20'}],
+ [{'start': '10.0.2.250', 'end': '10.0.3.5'}],
+ [{'start': '10.0.2.10', 'end': '10.0.2.5'}],
[{'start': '10.0.0.2', 'end': '10.0.0.3'},
{'start': '10.0.0.2', 'end': '10.0.0.3'}]]
tenant_id = network['network']['tenant_id']
return orig(s, instance)
mock.patch.object(orm.Session, 'add',
new=db_ref_err_for_ipalloc).start()
- mock.patch.object(non_ipam.IpamNonPluggableBackend,
+ mock.patch.object(db_base_plugin_common.DbBasePluginCommon,
'_get_subnet',
return_value=mock.Mock()).start()
# Add an IPv6 auto-address subnet to the network
import mock
import netaddr
+import webob.exc
+from oslo_config import cfg
from oslo_utils import uuidutils
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
+from neutron.db import ipam_backend_mixin
from neutron.db import ipam_pluggable_backend
from neutron.ipam import requests as ipam_req
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_base
+class UseIpamMixin(object):
+
+ def setUp(self):
+ cfg.CONF.set_override("ipam_driver", 'internal')
+ super(UseIpamMixin, self).setUp()
+
+
+class TestIpamHTTPResponse(UseIpamMixin, test_db_base.TestV2HTTPResponse):
+ pass
+
+
+class TestIpamPorts(UseIpamMixin, test_db_base.TestPortsV2):
+ pass
+
+
+class TestIpamNetworks(UseIpamMixin, test_db_base.TestNetworksV2):
+ pass
+
+
+class TestIpamSubnets(UseIpamMixin, test_db_base.TestSubnetsV2):
+ pass
+
+
+class TestIpamSubnetPool(UseIpamMixin, test_db_base.TestSubnetPoolsV2):
+ pass
+
+
class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
def setUp(self):
+ cfg.CONF.set_override("ipam_driver", 'internal')
super(TestDbBasePluginIpam, self).setUp()
self.tenant_id = uuidutils.generate_uuid()
self.subnet_id = uuidutils.generate_uuid()
mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend()
return mocks
+ def _prepare_mocks_with_pool_mock(self, pool_mock):
+ mocks = self._prepare_mocks()
+ pool_mock.get_instance.return_value = mocks['driver']
+ return mocks
+
def _get_allocate_mock(self, auto_ip='10.0.0.2',
fail_ip='127.0.0.1',
error_message='SomeError'):
self._validate_allocate_calls(ips[:-1], mocks)
# Deallocate should be called for the first ip only
mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_create_subnet_over_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ cidr = '192.168.0.0/24'
+ allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}]
+ with self.subnet(allocation_pools=allocation_pools,
+ cidr=cidr):
+ pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
+ assert mocks['driver'].allocate_subnet.called
+ request = mocks['driver'].allocate_subnet.call_args[0][0]
+ self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
+ self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_create_subnet_over_ipam_with_rollback(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ mocks['driver'].allocate_subnet.side_effect = ValueError
+ cidr = '10.0.2.0/24'
+ with self.network() as network:
+ self._create_subnet(self.fmt, network['network']['id'],
+ cidr, expected_res_status=500)
+
+ pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
+ assert mocks['driver'].allocate_subnet.called
+ request = mocks['driver'].allocate_subnet.call_args[0][0]
+ self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
+ self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
+ # Verify no subnet was created for network
+ req = self.new_show_request('networks', network['network']['id'])
+ res = req.get_response(self.api)
+ net = self.deserialize(self.fmt, res)
+ self.assertEqual(0, len(net['network']['subnets']))
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_ipam_subnet_deallocated_if_create_fails(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ cidr = '10.0.2.0/24'
+ with mock.patch.object(
+ ipam_backend_mixin.IpamBackendMixin, '_save_subnet',
+ side_effect=ValueError), self.network() as network:
+ self._create_subnet(self.fmt, network['network']['id'],
+ cidr, expected_res_status=500)
+ pool_mock.get_instance.assert_any_call(None, mock.ANY)
+ self.assertEqual(2, pool_mock.get_instance.call_count)
+ assert mocks['driver'].allocate_subnet.called
+ request = mocks['driver'].allocate_subnet.call_args[0][0]
+ self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
+ self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
+ # Verify remove ipam subnet was called
+ mocks['driver'].remove_subnet.assert_called_once_with(
+ self.subnet_id)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_update_subnet_over_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ cidr = '10.0.0.0/24'
+ allocation_pools = [{'start': '10.0.0.2', 'end': '10.0.0.254'}]
+ with self.subnet(allocation_pools=allocation_pools,
+ cidr=cidr) as subnet:
+ data = {'subnet': {'allocation_pools': [
+ {'start': '10.0.0.10', 'end': '10.0.0.20'},
+ {'start': '10.0.0.30', 'end': '10.0.0.40'}]}}
+ req = self.new_update_request('subnets', data,
+ subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_code, 200)
+
+ pool_mock.get_instance.assert_any_call(None, mock.ANY)
+ self.assertEqual(2, pool_mock.get_instance.call_count)
+ assert mocks['driver'].update_subnet.called
+ request = mocks['driver'].update_subnet.call_args[0][0]
+ self.assertEqual(ipam_req.SpecificSubnetRequest, type(request))
+ self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
+
+ ip_ranges = [netaddr.IPRange(p['start'],
+ p['end']) for p in data['subnet']['allocation_pools']]
+ self.assertEqual(ip_ranges, request.allocation_pools)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_delete_subnet_over_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ gateway_ip = '10.0.0.1'
+ cidr = '10.0.0.0/24'
+ res = self._create_network(fmt=self.fmt, name='net',
+ admin_state_up=True)
+ network = self.deserialize(self.fmt, res)
+ subnet = self._make_subnet(self.fmt, network, gateway_ip,
+ cidr, ip_version=4)
+ req = self.new_delete_request('subnets', subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
+
+ pool_mock.get_instance.assert_any_call(None, mock.ANY)
+ self.assertEqual(2, pool_mock.get_instance.call_count)
+ mocks['driver'].remove_subnet.assert_called_once_with(
+ subnet['subnet']['id'])
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_delete_subnet_over_ipam_with_rollback(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ mocks['driver'].remove_subnet.side_effect = ValueError
+ gateway_ip = '10.0.0.1'
+ cidr = '10.0.0.0/24'
+ res = self._create_network(fmt=self.fmt, name='net',
+ admin_state_up=True)
+ network = self.deserialize(self.fmt, res)
+ subnet = self._make_subnet(self.fmt, network, gateway_ip,
+ cidr, ip_version=4)
+ req = self.new_delete_request('subnets', subnet['subnet']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int, webob.exc.HTTPServerError.code)
+
+ pool_mock.get_instance.assert_any_call(None, mock.ANY)
+ self.assertEqual(2, pool_mock.get_instance.call_count)
+ mocks['driver'].remove_subnet.assert_called_once_with(
+ subnet['subnet']['id'])
+ # Verify subnet was recreated after failed ipam call
+ subnet_req = self.new_show_request('subnets',
+ subnet['subnet']['id'])
+ raw_res = subnet_req.get_response(self.api)
+ sub_res = self.deserialize(self.fmt, raw_res)
+ self.assertIn(sub_res['subnet']['cidr'], cidr)
+ self.assertIn(sub_res['subnet']['gateway_ip'],
+ gateway_ip)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_create_port_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ auto_ip = '10.0.0.2'
+ expected_calls = [{'ip_address': ''}]
+ mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
+ auto_ip=auto_ip)
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ ips = port['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(ips[0]['ip_address'], auto_ip)
+ self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
+ self._validate_allocate_calls(expected_calls, mocks)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_create_port_ipam_with_rollback(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ mocks['subnet'].allocate.side_effect = ValueError
+ with self.network() as network:
+ with self.subnet(network=network):
+ net_id = network['network']['id']
+ data = {
+ 'port': {'network_id': net_id,
+ 'tenant_id': network['network']['tenant_id']}}
+ port_req = self.new_create_request('ports', data)
+ res = port_req.get_response(self.api)
+ self.assertEqual(res.status_int,
+ webob.exc.HTTPServerError.code)
+
+ # verify no port left after failure
+ req = self.new_list_request('ports', self.fmt,
+ "network_id=%s" % net_id)
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ self.assertEqual(0, len(res['ports']))
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_update_port_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ auto_ip = '10.0.0.2'
+ new_ip = '10.0.0.15'
+ expected_calls = [{'ip_address': ip} for ip in ['', new_ip]]
+ mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
+ auto_ip=auto_ip)
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ ips = port['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(ips[0]['ip_address'], auto_ip)
+ # Update port with another new ip
+ data = {"port": {"fixed_ips": [{
+ 'subnet_id': subnet['subnet']['id'],
+ 'ip_address': new_ip}]}}
+ req = self.new_update_request('ports', data,
+ port['port']['id'])
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ ips = res['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(new_ip, ips[0]['ip_address'])
+
+ # Allocate should be called for the first two networks
+ self._validate_allocate_calls(expected_calls, mocks)
+ # Deallocate should be called for the first ip only
+ mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
+
+ @mock.patch('neutron.ipam.driver.Pool')
+ def test_delete_port_ipam(self, pool_mock):
+ mocks = self._prepare_mocks_with_pool_mock(pool_mock)
+ auto_ip = '10.0.0.2'
+ mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
+ auto_ip=auto_ip)
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ ips = port['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(ips[0]['ip_address'], auto_ip)
+ req = self.new_delete_request('ports', port['port']['id'])
+ res = req.get_response(self.api)
+
+ self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
+ mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
+
+ def test_recreate_port_ipam(self):
+ ip = '10.0.0.2'
+ with self.subnet() as subnet:
+ with self.port(subnet=subnet) as port:
+ ips = port['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(ips[0]['ip_address'], ip)
+ req = self.new_delete_request('ports', port['port']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int, webob.exc.HTTPNoContent.code)
+ with self.port(subnet=subnet, fixed_ips=ips) as port:
+ ips = port['port']['fixed_ips']
+ self.assertEqual(1, len(ips))
+ self.assertEqual(ips[0]['ip_address'], ip)
id=self.ipam_subnet_id).all()
self.assertEqual(1, len(subnets))
- def test_associate_neutron_id(self):
- self.subnet_manager.associate_neutron_id(self.ctx.session,
- 'test-id')
- subnet = self.ctx.session.query(db_models.IpamSubnet).filter_by(
- id=self.ipam_subnet_id).first()
- self.assertEqual('test-id', subnet['neutron_subnet_id'])
+ def test_remove(self):
+ count = db_api.IpamSubnetManager.delete(self.ctx.session,
+ self.neutron_subnet_id)
+ self.assertEqual(1, count)
+ subnets = self.ctx.session.query(db_models.IpamSubnet).filter_by(
+ id=self.ipam_subnet_id).all()
+ self.assertEqual(0, len(subnets))
+
+ def test_remove_non_existent_subnet(self):
+ count = db_api.IpamSubnetManager.delete(self.ctx.session,
+ 'non-existent')
+ self.assertEqual(0, count)
def _create_pools(self, pools):
db_pools = []
def test_update_subnet_pools(self):
cidr = '10.0.0.0/24'
subnet, subnet_req = self._prepare_specific_subnet_request(cidr)
- ipam_subnet = self.ipam_pool.allocate_subnet(subnet_req)
- ipam_subnet.associate_neutron_subnet(subnet['id'])
+ self.ipam_pool.allocate_subnet(subnet_req)
allocation_pools = [netaddr.IPRange('10.0.0.100', '10.0.0.150'),
netaddr.IPRange('10.0.0.200', '10.0.0.250')]
update_subnet_req = ipam_req.SpecificSubnetRequest(
def test_get_subnet(self):
cidr = '10.0.0.0/24'
subnet, subnet_req = self._prepare_specific_subnet_request(cidr)
- ipam_subnet = self.ipam_pool.allocate_subnet(subnet_req)
- ipam_subnet.associate_neutron_subnet(subnet['id'])
+ self.ipam_pool.allocate_subnet(subnet_req)
# Retrieve the subnet
ipam_subnet = self.ipam_pool.get_subnet(subnet['id'])
self._verify_ipam_subnet_details(
self.ipam_pool.get_subnet,
'boo')
+ def test_remove_ipam_subnet(self):
+ cidr = '10.0.0.0/24'
+ subnet, subnet_req = self._prepare_specific_subnet_request(cidr)
+ self.ipam_pool.allocate_subnet(subnet_req)
+ # Remove ipam subnet by neutron subnet id
+ self.ipam_pool.remove_subnet(subnet['id'])
+
+ def test_remove_non_existent_subnet_fails(self):
+ self.assertRaises(n_exc.SubnetNotFound,
+ self.ipam_pool.remove_subnet,
+ 'non-existent-id')
+
+ def test_get_details_for_invalid_subnet_id_fails(self):
+ cidr = '10.0.0.0/24'
+ subnet_req = ipam_req.SpecificSubnetRequest(
+ self._tenant_id,
+ 'non-existent-id',
+ cidr)
+ self.ipam_pool.allocate_subnet(subnet_req)
+ # Neutron subnet does not exist, so get_subnet should fail
+ self.assertRaises(n_exc.SubnetNotFound,
+ self.ipam_pool.get_subnet,
+ 'non-existent-id')
+
class TestNeutronDbIpamSubnet(testlib_api.SqlTestCase,
TestNeutronDbIpamMixin):
gateway_ip=subnet['gateway_ip'],
allocation_pools=allocation_pool_ranges)
ipam_subnet = self.ipam_pool.allocate_subnet(subnet_req)
- ipam_subnet.associate_neutron_subnet(subnet['id'])
return ipam_subnet, subnet
def setUp(self):
subnet = self._create_subnet(
self.plugin, self.ctx, self.net_id, cidr)
subnet_req = ipam_req.SpecificSubnetRequest(
- 'tenant_id', subnet, cidr, gateway_ip=subnet['gateway_ip'])
+ 'tenant_id', subnet['id'], cidr, gateway_ip=subnet['gateway_ip'])
ipam_subnet = self.ipam_pool.allocate_subnet(subnet_req)
with self.ctx.session.begin():
ranges = ipam_subnet._allocate_specific_ip(
# This test instead might be made to pass, but for the wrong reasons!
pass
- def _test_allocate_subnet(self, subnet_id):
- subnet_req = ipam_req.SpecificSubnetRequest(
- 'tenant_id', subnet_id, '192.168.0.0/24')
- return self.ipam_pool.allocate_subnet(subnet_req)
-
def test_allocate_subnet_for_non_existent_subnet_pass(self):
- # This test should pass because neutron subnet is not checked
- # until associate neutron subnet step
+ # This test should pass because ipam subnet is no longer
+ # have foreign key relationship with neutron subnet.
+ # Creating ipam subnet before neutron subnet is a valid case.
subnet_req = ipam_req.SpecificSubnetRequest(
'tenant_id', 'meh', '192.168.0.0/24')
self.ipam_pool.allocate_subnet(subnet_req)
-
- def test_associate_neutron_subnet(self):
- ipam_subnet, subnet = self._create_and_allocate_ipam_subnet(
- '192.168.0.0/24', ip_version=4)
- details = ipam_subnet.get_details()
- self.assertEqual(subnet['id'], details.subnet_id)
-
- def test_associate_non_existing_neutron_subnet_fails(self):
- subnet_req = ipam_req.SpecificSubnetRequest(
- 'tenant_id', 'meh', '192.168.0.0/24')
- ipam_subnet = self.ipam_pool.allocate_subnet(subnet_req)
- self.assertRaises(n_exc.SubnetNotFound,
- ipam_subnet.associate_neutron_subnet,
- 'meh')
class ContrailPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
_plugin_name = ('%s.NeutronPluginContrailCoreV2' % CONTRAIL_PKG_PATH)
+ _fetch = ('neutron.ipam.drivers.neutrondb_ipam.driver.NeutronDbSubnet'
+ '._fetch_subnet')
def setUp(self, plugin=None, ext_mgr=None):
if 'v6' in self._testMethodName:
self.skipTest("OpenContrail Plugin does not support subnet pools.")
cfg.CONF.keystone_authtoken = KeyStoneInfo()
mock.patch('requests.post').start().side_effect = FAKE_SERVER.request
+ mock.patch(self._fetch).start().side_effect = FAKE_SERVER._get_subnet
super(ContrailPluginTestCase, self).setUp(self._plugin_name)
neutron.tests.unit.scheduler.test_dhcp_agent_scheduler \
neutron.tests.unit.db.test_ipam_backend_mixin \
neutron.tests.unit.db.test_l3_dvr_db \
- neutron.tests.unit.db.test_ipam_pluggable_backend \
neutron.tests.unit.db.test_migration \
neutron.tests.unit.db.test_agents_db \
neutron.tests.unit.db.test_dvr_mac_db \
Code Review / openstack-build / neutron-build.git / commitdiff