from neutron.db import model_base
from neutron.db import models_v2
from neutron.extensions import external_net
+from neutron import manager
+from neutron.plugins.common import constants as service_constants
DEVICE_OWNER_ROUTER_GW = l3_constants.DEVICE_OWNER_ROUTER_GW
network_id=net_id).delete()
net_data[external_net.EXTERNAL] = False
+ def _process_l3_delete(self, context, network_id):
+ l3plugin = manager.NeutronManager.get_service_plugins().get(
+ service_constants.L3_ROUTER_NAT)
+ if l3plugin:
+ l3plugin.delete_disassociated_floatingips(context, network_id)
+
def _filter_nets_l3(self, context, nets, filters):
vals = filters and filters.get(external_net.EXTERNAL, [])
if not vals:
marker_obj=marker_obj,
page_reverse=page_reverse)
+ def delete_disassociated_floatingips(self, context, network_id):
+ query = self._model_query(context, FloatingIP)
+ query = query.filter_by(floating_network_id=network_id,
+ fixed_port_id=None,
+ router_id=None)
+ for fip in query:
+ self.delete_floatingip(context, fip.id)
+
def get_floatingips_count(self, context, filters=None):
return self._get_collection_count(context, FloatingIP,
filters=filters)
# Validate args
orig_net = super(NeutronRestProxyV2, self).get_network(context, net_id)
-
- filter = {'network_id': [net_id]}
- ports = self.get_ports(context, filters=filter)
-
- # check if there are any tenant owned ports in-use
- auto_delete_port_owners = db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS
- only_auto_del = all(p['device_owner'] in auto_delete_port_owners
- for p in ports)
-
- if not only_auto_del:
- raise exceptions.NetworkInUse(net_id=net_id)
with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, net_id)
ret_val = super(NeutronRestProxyV2, self).delete_network(context,
net_id)
self._send_delete_network(orig_net, context)
port_id)
self._send_floatingip_update(context)
+ # overriding method from l3_db as original method calls
+ # self.delete_floatingip() which in turn calls self.delete_port() which
+ # is locked with 'bsn-port-barrier'
+ def delete_disassociated_floatingips(self, context, network_id):
+ query = self._model_query(context, l3_db.FloatingIP)
+ query = query.filter_by(floating_network_id=network_id,
+ fixed_port_id=None,
+ router_id=None)
+ for fip in query:
+ context.session.delete(fip)
+ self._delete_port(context.elevated(), fip['floating_port_id'])
+
def _send_floatingip_update(self, context):
try:
ext_net_id = self.get_external_network_id(context)
"""
with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, net_id)
result = super(BrocadePluginV2, self).delete_network(context,
net_id)
# we must delete all ports in db first (foreign key constraint)
# the network_binding record is deleted via cascade from
# the network record, so explicit removal is not necessary
self._send_delete_network_request(context, network)
+
+ self._process_l3_delete(context, id)
super(N1kvNeutronPluginV2, self).delete_network(context, id)
LOG.debug(_("Deleted network: %s"), id)
session = context.session
with session.begin(subtransactions=True):
binding = self._db.get_network_binding(session, id)
+ self._process_l3_delete(context, id)
super(HyperVNeutronPlugin, self).delete_network(context, id)
p = self._network_providers_map[binding.network_type]
p.delete_network(session, binding)
@_ha
def delete_network(self, context, id):
LOG.debug(_("Delete network in progress: %s"), id)
- super(SdnvePluginV2, self).delete_network(context, id)
+ session = context.session
+
+ with session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ super(SdnvePluginV2, self).delete_network(context, id)
(res, data) = self.sdnve_client.sdnve_delete('network', id)
if res not in constants.HTTP_ACCEPTABLE:
session = context.session
with session.begin(subtransactions=True):
binding = db.get_network_binding(session, id)
+ self._process_l3_delete(context, id)
super(LinuxBridgePluginV2, self).delete_network(context, id)
if binding.vlan_id != constants.LOCAL_VLAN_ID:
db.release_network(session, binding.physical_network,
LOG.debug(_("MidonetPluginV2.delete_network called: id=%r"), id)
self.client.delete_bridge(id)
try:
- super(MidonetPluginV2, self).delete_network(context, id)
+ with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ super(MidonetPluginV2, self).delete_network(context, id)
except Exception:
with excutils.save_and_reraise_exception():
LOG.error(_('Failed to delete neutron db, while Midonet '
while True:
try:
with session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+
# Get ports to auto-delete.
ports = (session.query(models_v2.Port).
enable_eagerloads(False).
session = context.session
with session.begin(subtransactions=True):
binding = db.get_network_binding(session, net_id)
+ self._process_l3_delete(context, net_id)
super(MellanoxEswitchPlugin, self).delete_network(context,
net_id)
if binding.segmentation_id != constants.LOCAL_VLAN_ID:
tenant_id = net_db['tenant_id']
ports = self.get_ports(context, filters={'network_id': [id]})
- # check if there are any tenant owned ports in-use
+ # check if there are any tenant owned ports in-use;
+ # consider ports owned by floating ips as auto_delete as if there are
+ # no other tenant owned ports, those floating ips are disassociated
+ # and will be auto deleted with self._process_l3_delete()
only_auto_del = all(p['device_owner'] in
- db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS
+ db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS or
+ p['device_owner'] == const.DEVICE_OWNER_FLOATINGIP
for p in ports)
if not only_auto_del:
raise n_exc.NetworkInUse(net_id=id)
+ self._process_l3_delete(context, id)
+
# Make sure auto-delete ports on OFC are deleted.
# If an error occurs during port deletion,
# delete_network will be aborted.
- for port in ports:
+ for port in [p for p in ports if p['device_owner']
+ in db_base_plugin_v2.AUTO_DELETE_PORT_OWNERS]:
port = self.deactivate_port(context, port)
# delete all packet_filters of the network from the controller
return net
def delete_network(self, context, id):
- filter = {'network_id': [id]}
- subnets = self.get_subnets(context, filters=filter)
- for subnet in subnets:
- self.delete_subnet(context, subnet['id'])
- super(NuagePlugin, self).delete_network(context, id)
+ with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ filter = {'network_id': [id]}
+ subnets = self.get_subnets(context, filters=filter)
+ for subnet in subnets:
+ self.delete_subnet(context, subnet['id'])
+ super(NuagePlugin, self).delete_network(context, id)
def _get_net_partition_for_subnet(self, context, subnet):
subn = subnet['subnet']
#get all the subnets under the network to delete them
subnets = self._get_subnets_by_network(context, net_id)
+ self._process_l3_delete(context, net_id)
super(OneConvergencePluginV2, self).delete_network(context,
net_id)
session = context.session
with session.begin(subtransactions=True):
binding = ovs_db_v2.get_network_binding(session, id)
+ self._process_l3_delete(context, id)
super(OVSNeutronPluginV2, self).delete_network(context, id)
if binding.network_type in constants.TUNNEL_NETWORK_TYPES:
ovs_db_v2.release_tunnel(session, binding.segmentation_id,
self).get_network(context, net_id)
with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, net_id)
# Plugin DB - Network Delete
super(NeutronPluginPLUMgridV2, self).delete_network(context,
net_id)
session = context.session
with session.begin(subtransactions=True):
self.tunnel_key.delete(session, id)
+ self._process_l3_delete(context, id)
super(RyuNeutronPluginV2, self).delete_network(context, id)
def create_port(self, context, port):
if not external:
lswitch_ids = nsx_utils.get_nsx_switch_ids(
context.session, self.cluster, id)
- super(NsxPluginV2, self).delete_network(context, id)
+ with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ super(NsxPluginV2, self).delete_network(context, id)
+
# clean up network owned ports
for port in router_iface_ports:
try:
return net
def delete_network(self, context, id):
- return super(Fake1, self).delete_network(context, id)
+ session = context.session
+ with session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ return super(Fake1, self).delete_network(context, id)
def create_port(self, context, port):
port = super(Fake1, self).create_port(context, port)
import contextlib
import itertools
+import mock
import testtools
from webob import exc
self.assertEqual(ext_net['network'][external_net.EXTERNAL],
True)
+ def test_delete_network_check_disassociated_floatingips(self):
+ with mock.patch.object(NeutronManager,
+ 'get_service_plugins') as srv_plugins:
+ l3_mock = mock.Mock()
+ srv_plugins.return_value = {'L3_ROUTER_NAT': l3_mock}
+ with self.network(do_delete=False) as net:
+ req = self.new_delete_request('networks', net['network']['id'])
+ res = req.get_response(self.api)
+ self.assertEqual(res.status_int, exc.HTTPNoContent.code)
+ (l3_mock.delete_disassociated_floatingips
+ .assert_called_once_with(mock.ANY, net['network']['id']))
+
class ExtNetDBTestCaseXML(ExtNetDBTestCase):
fmt = 'xml'
self._process_l3_update(context, net, network['network'])
return net
+ def delete_network(self, context, id):
+ with context.session.begin(subtransactions=True):
+ self._process_l3_delete(context, id)
+ super(TestL3NatBasePlugin, self).delete_network(context, id)
+
def delete_port(self, context, id, l3_port_check=True):
plugin = NeutronManager.get_service_plugins().get(
service_constants.L3_ROUTER_NAT)
s['subnet']['id'],
None)
+ def test_delete_ext_net_with_disassociated_floating_ips(self):
+ with self.network() as net:
+ net_id = net['network']['id']
+ self._set_net_external(net_id)
+ with self.subnet(network=net, do_delete=False):
+ self._make_floatingip(self.fmt, net_id)
+
class L3AgentDbTestCaseBase(L3NatTestCaseMixin):
Code Review / openstack-build / neutron-build.git / commitdiff