Merge "NSX-V3: Prevent deletion of router gw/interface used by LB"

vmware_nsx/services/lbaas/nsx_v3/lb_driver_v2.py

@@ -14,9 +14,10 @@
 #    under the License.
 
 from neutron_lib.callbacks import events
-from neutron_lib.callbacks import exceptions as nc_exc
 from neutron_lib.callbacks import registry
 from neutron_lib.callbacks import resources
+from neutron_lib import exceptions as n_exc
+from neutron_lib.plugins import constants as plugin_const
 from oslo_log import helpers as log_helpers
 from oslo_log import log as logging
 
@@ -64,30 +65,77 @@ class EdgeLoadbalancerDriverV2(object):
 
     def _subscribe_router_delete_callback(self):
         # Check if there is any LB attachment for the NSX router.
-        # This callback is subscribed here to prevent router deletion
+        # This callback is subscribed here to prevent router/GW/interface
-        # if it still has LB service attached to it.
+        # deletion if it still has LB service attached to it.
         registry.subscribe(self._check_lb_service_on_router,
                            resources.ROUTER, events.BEFORE_DELETE)
+        registry.subscribe(self._check_lb_service_on_router,
+                           resources.ROUTER_GATEWAY, events.BEFORE_DELETE)
+        registry.subscribe(self._check_lb_service_on_router_interface,
+                           resources.ROUTER_INTERFACE, events.BEFORE_DELETE)
 
     def _unsubscribe_router_delete_callback(self):
         registry.unsubscribe(self._check_lb_service_on_router,
                              resources.ROUTER, events.BEFORE_DELETE)
+        registry.unsubscribe(self._check_lb_service_on_router,
+                             resources.ROUTER_GATEWAY, events.BEFORE_DELETE)
+        registry.unsubscribe(self._check_lb_service_on_router_interface,
+                             resources.ROUTER_INTERFACE, events.BEFORE_DELETE)
+
+    def _get_lb_ports(self, context, subnet_ids):
+        dev_owner = 'neutron:' + plugin_const.LOADBALANCERV2
+        filters = {'device_owner': [dev_owner],
+                   'fixed_ips': {'subnet_id': subnet_ids}}
+        return self.loadbalancer.core_plugin.get_ports(
+            context, filters=filters)
 
     def _check_lb_service_on_router(self, resource, event, trigger,
                                     **kwargs):
-        """Check if there is any lb service on nsx router"""
+        """Prevent removing a router GW or deleting a router used by LB"""
+        router_id = kwargs.get('router_id')
+        context = kwargs['context']
+        nsx_router_id = nsx_db.get_nsx_router_id(kwargs['context'].session,
+                                                 router_id)
+        if not nsx_router_id:
+            # Skip non-v3 routers (could be a V router in case of TVD plugin)
+            return
+        nsxlib = self.loadbalancer.core_plugin.nsxlib
+        service_client = nsxlib.load_balancer.service
+        # Check if there is any lb service on nsx router
+        lb_service = service_client.get_router_lb_service(nsx_router_id)
+        if lb_service:
+            msg = _('Cannot delete a %s as it still has lb service '
+                    'attachment') % resource
+            raise n_exc.BadRequest(resource='lbaas-lb', msg=msg)
+
+        # Also check if there are any loadbalancers attached to this router
+        # subnets
+        router_subnets = self.loadbalancer.core_plugin._find_router_subnets(
+            context.elevated(), router_id)
+        subnet_ids = [subnet['id'] for subnet in router_subnets]
+        if self._get_lb_ports(context.elevated(), subnet_ids):
+            msg = (_('Cannot delete a %s as it used by a loadbalancer') %
+                   resource)
+            raise n_exc.BadRequest(resource='lbaas-lb', msg=msg)
+
+    def _check_lb_service_on_router_interface(self, *args, **kwargs):
+        # Prevent removing the interface of an LB subnet from a router
+        router_id = kwargs.get('router_id')
+        subnet_id = kwargs.get('subnet_id')
+        if not router_id or not subnet_id:
+            return
 
         nsx_router_id = nsx_db.get_nsx_router_id(kwargs['context'].session,
                                                  kwargs['router_id'])
         if not nsx_router_id:
+            # Skip non-v3 routers (could be a V router in case of TVD plugin)
             return
-        nsxlib = self.loadbalancer.core_plugin.nsxlib
+
-        service_client = nsxlib.load_balancer.service
+        # get LB ports and check if any loadbalancer is using this subnet
-        lb_service = service_client.get_router_lb_service(nsx_router_id)
+        if self._get_lb_ports(kwargs['context'].elevated(), [subnet_id]):
-        if lb_service:
+            msg = _('Cannot delete a router interface as it used by a '
-            msg = _('Cannot delete router as it still has lb service '
+                    'loadbalancer')
-                    'attachment')
+            raise n_exc.BadRequest(resource='lbaas-lb', msg=msg)
-            raise nc_exc.CallbackFailure(msg)
 
 
 class DummyLoadbalancerDriverV2(object):

vmware_nsx/tests/unit/nsx_v3/test_plugin.py

@@ -1406,10 +1406,14 @@ class L3NatTest(test_l3_plugin.L3BaseForIntTests, NsxV3PluginTestCaseMixin,
                                              new=lambda v: True)
         mock_nsx_version.start()
         # Make sure the LB callback is not called on router deletion
-        self.lb_mock = mock.patch(
+        self.lb_mock1 = mock.patch(
             "vmware_nsx.services.lbaas.nsx_v3.lb_driver_v2."
             "EdgeLoadbalancerDriverV2._check_lb_service_on_router")
-        self.lb_mock.start()
+        self.lb_mock1.start()
+        self.lb_mock2 = mock.patch(
+            "vmware_nsx.services.lbaas.nsx_v3.lb_driver_v2."
+            "EdgeLoadbalancerDriverV2._check_lb_service_on_router_interface")
+        self.lb_mock2.start()
 
         super(L3NatTest, self).setUp(
             plugin=plugin, ext_mgr=ext_mgr, service_plugins=service_plugins)
@@ -1495,19 +1499,23 @@ class TestL3NatTestCase(L3NatTest,
         self.skipTest('not supported')
 
     def test_router_delete_with_lb_service(self):
-        self.lb_mock.stop()
+        self.lb_mock1.stop()
+        self.lb_mock2.stop()
         # Create the LB object - here the delete callback is registered
         lb_driver = lb_driver_v2.EdgeLoadbalancerDriverV2()
         with self.router() as router:
             with mock.patch('vmware_nsxlib.v3.load_balancer.Service.'
-                            'get_router_lb_service'):
+                            'get_router_lb_service'),\
+                mock.patch('vmware_nsx.db.db.get_nsx_router_id',
+                           return_value=1):
                 self.assertRaises(nc_exc.CallbackFailure,
                                   self.plugin_instance.delete_router,
                                   context.get_admin_context(),
                                   router['router']['id'])
         # Unregister callback
         lb_driver._unsubscribe_router_delete_callback()
-        self.lb_mock.start()
+        self.lb_mock1.start()
+        self.lb_mock2.start()
 
     def test_multiple_subnets_on_different_routers(self):
         with self.network() as network: