NSXv3: Prevent router deletion if it has lb attachment

Use Neutron callback to prevent router deletion if router still
has lb service attachment.

Co-Authored-By: Gary Kotton <gkotton@vmware.com>
Change-Id: I38bd169507d7aa9cbfbf691fa63060701baf77c0

vmware_nsx/plugins/nsx_v3/plugin.py

@@ -2925,8 +2925,6 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
         return self.get_router(context, router['id'])
 
     def delete_router(self, context, router_id):
-        # TODO(tongl) Prevent router deletion if router still has load
-        # balancer attachment. Raise exception if router has lb attachment.
         if not cfg.CONF.nsx_v3.native_dhcp_metadata:
             nsx_rpc.handle_router_metadata_access(self, context, router_id,
                                                   interface=None)

vmware_nsx/services/lbaas/nsx_v3/lb_driver_v2.py

@@ -13,9 +13,15 @@
 #    License for the specific language governing permissions and limitations
 #    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 oslo_log import helpers as log_helpers
 from oslo_log import log as logging
 
+from vmware_nsx._i18n import _
+from vmware_nsx.db import db as nsx_db
 from vmware_nsx.services.lbaas.nsx_v3 import healthmonitor_mgr as hm_mgr
 from vmware_nsx.services.lbaas.nsx_v3 import l7policy_mgr
 from vmware_nsx.services.lbaas.nsx_v3 import l7rule_mgr
@@ -54,6 +60,32 @@ class EdgeLoadbalancerDriverV2(object):
         self.healthmonitor = hm_mgr.EdgeHealthMonitorManager()
         self.l7policy = l7policy_mgr.EdgeL7PolicyManager()
         self.l7rule = l7rule_mgr.EdgeL7RuleManager()
+        self._subscribe_router_delete_callback()
+
+    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
+        # if it still has LB service attached to it.
+        registry.subscribe(self._check_lb_service_on_router,
+                           resources.ROUTER, events.BEFORE_DELETE)
+
+    def _unsubscribe_router_delete_callback(self):
+        registry.unsubscribe(self._check_lb_service_on_router,
+                             resources.ROUTER, events.BEFORE_DELETE)
+
+    def _check_lb_service_on_router(self, resource, event, trigger,
+                                    **kwargs):
+        """Check if there is any lb service on nsx router"""
+
+        nsx_router_id = nsx_db.get_nsx_router_id(kwargs['context'].session,
+                                                 kwargs['router_id'])
+        nsxlib = self.loadbalancer.core_plugin.nsxlib
+        service_client = nsxlib.load_balancer.service
+        lb_service = service_client.get_router_lb_service(nsx_router_id)
+        if lb_service:
+            msg = _('Cannot delete router as it still has lb service '
+                    'attachment')
+            raise nc_exc.CallbackFailure(msg)
 
 
 class DummyLoadbalancerDriverV2(object):

vmware_nsx/tests/unit/nsx_v3/test_plugin.py

@@ -40,6 +40,7 @@ from neutron.tests.unit.scheduler \
 from neutron_lib.api.definitions import address_scope as addr_apidef
 from neutron_lib.api.definitions import portbindings
 from neutron_lib.api.definitions import provider_net as pnet
+from neutron_lib.callbacks import exceptions as nc_exc
 from neutron_lib import constants
 from neutron_lib import context
 from neutron_lib import exceptions as n_exc
@@ -50,6 +51,7 @@ from oslo_utils import uuidutils
 from vmware_nsx.api_client import exception as api_exc
 from vmware_nsx.common import utils
 from vmware_nsx.plugins.nsx_v3 import plugin as nsx_plugin
+from vmware_nsx.services.lbaas.nsx_v3 import lb_driver_v2
 from vmware_nsx.tests import unit as vmware
 from vmware_nsx.tests.unit.extensions import test_metadata
 from vmware_nsxlib.tests.unit.v3 import mocks as nsx_v3_mocks
@@ -162,6 +164,10 @@ def _mock_nsx_backend_calls():
         "vmware_nsxlib.v3.NsxLib.get_version",
         return_value='1.1.0').start()
 
+    mock.patch(
+        "vmware_nsxlib.v3.load_balancer.Service.get_router_lb_service",
+        return_value=None).start()
+
 
 class NsxV3PluginTestCaseMixin(test_plugin.NeutronDbPluginV2TestCase,
                                nsxlib_testcase.NsxClientTestCase):
@@ -748,6 +754,19 @@ class TestL3NatTestCase(L3NatTest,
     def test_floatingip_update_subnet_gateway_disabled(self):
         self.skipTest('not supported')
 
+    def test_router_delete_with_lb_service(self):
+        # 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'):
+                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()
+
     def test_multiple_subnets_on_different_routers(self):
         with self.network() as network:
             with self.subnet(network=network) as s1,\