Prevent deletion of a subnet with lbaas v1 pool

When using LBaaS and trying to delete a subnet, neutron has no way of
knowing if the subnet is associated to some pool. As a result, the
subnet is deleted but the pool remains associated to the (now
nonexistent) subnet_id.  This patch adds a check in LBaaS side to
prevent such a case, using the callbacks system.

Closes-Bug: #1413817
Change-Id: I1a8893453b0b623df792fdabcf540163007a3f33
Depends-on: I3d5e231b67c72ffd919c92d65b57da56c63e053c

neutron_lbaas/db/loadbalancer/loadbalancer_db.py

@@ -23,6 +23,7 @@ from neutron.db import common_db_mixin as base_db
 from neutron.db import model_base
 from neutron.db import models_v2
 from neutron.db import servicetype_db as st_db
+from neutron.i18n import _LE
 from neutron import manager
 from neutron.plugins.common import constants
 from oslo_db import exception
@@ -838,6 +839,13 @@ class LoadBalancerPluginDb(loadbalancer.LoadBalancerPluginBase,
                                     self._make_health_monitor_dict,
                                     filters=filters, fields=fields)
 
+    def check_subnet_in_use(self, context, subnet_id):
+        query = context.session.query(Pool).filter_by(subnet_id=subnet_id)
+        if query.count():
+            pool_id = query.one().id
+            raise n_exc.SubnetInUse(
+                reason=_LE("Subnet is used by loadbalancer pool %s") % pool_id)
+
 
 def _prevent_lbaas_port_delete_callback(resource, event, trigger, **kwargs):
     context = kwargs['context']
@@ -847,3 +855,20 @@ def _prevent_lbaas_port_delete_callback(resource, event, trigger, **kwargs):
                          constants.LOADBALANCER)
     if lbaasplugin and port_check:
         lbaasplugin.prevent_lbaas_port_deletion(context, port_id)
+
+
+def is_subnet_in_use_callback(resource, event, trigger, **kwargs):
+    service = manager.NeutronManager.get_service_plugins().get(
+        constants.LOADBALANCER)
+    if service:
+        context = kwargs.get('context')
+        subnet_id = kwargs.get('subnet_id')
+        service.check_subnet_in_use(context, subnet_id)
+
+
+def subscribe():
+    registry.subscribe(is_subnet_in_use_callback,
+                       resources.SUBNET, events.BEFORE_DELETE)
+
+
+subscribe()

neutron_lbaas/tests/unit/db/loadbalancer/test_db_loadbalancer.py

@@ -17,6 +17,9 @@ import contextlib
 
 import mock
 from neutron.api import extensions
+from neutron.callbacks import events
+from neutron.callbacks import registry
+from neutron.callbacks import resources
 from neutron.common import config
 from neutron.common import constants as n_constants
 from neutron.common import exceptions as n_exc
@@ -139,9 +142,9 @@ class LoadBalancerTestMixin(object):
         return vip_res
 
     def _create_pool(self, fmt, name, lb_method, protocol, admin_state_up,
-                     expected_res_status=None, **kwargs):
+                     subnet_id, expected_res_status=None, **kwargs):
         data = {'pool': {'name': name,
-                         'subnet_id': _subnet_id,
+                         'subnet_id': subnet_id,
                          'lb_method': lb_method,
                          'protocol': protocol,
                          'admin_state_up': admin_state_up,
@@ -228,14 +231,16 @@ class LoadBalancerTestMixin(object):
     @contextlib.contextmanager
     def pool(self, fmt=None, name='pool1', lb_method='ROUND_ROBIN',
              protocol='HTTP', admin_state_up=True, do_delete=True,
-             **kwargs):
+             subnet_id=None, **kwargs):
         if not fmt:
             fmt = self.fmt
+        subnet_id = subnet_id or _subnet_id
         res = self._create_pool(fmt,
                                 name,
                                 lb_method,
                                 protocol,
                                 admin_state_up,
+                                subnet_id,
                                 **kwargs)
         if res.status_int >= webob.exc.HTTPClientError.code:
             raise webob.exc.HTTPClientError(
@@ -767,6 +772,25 @@ class TestLoadBalancer(LoadBalancerPluginDbTestCase):
             req = self.new_delete_request('pools',
                                           pool['pool']['id'])
 
+    def test_delete_subnet_with_pool(self):
+        registry.subscribe(ldb.is_subnet_in_use_callback,
+                           resources.SUBNET, events.BEFORE_DELETE)
+
+        try:
+            with self.subnet() as subnet:
+                with self.pool(subnet_id=subnet['subnet']['id']):
+                    req = self.new_delete_request('subnets',
+                                                  subnet['subnet']['id'])
+                    res = req.get_response(self.api)
+
+                    self.assertTrue('NeutronError' in res.json)
+                    self.assertEqual('SubnetInUse',
+                                     res.json['NeutronError']['type'])
+                    self.assertEqual(409, res.status_code)
+        finally:
+            registry.unsubscribe(ldb.is_subnet_in_use_callback,
+                                 resources.SUBNET, events.BEFORE_DELETE)
+
     def test_show_pool(self):
         name = "pool1"
         keys = [('name', name),