port_forwarding: extend support for OVN usage

This commit adds possibility to configure fip port_forwarding
service plugin and l3 extension with devstack plugin for OVN.

Since OVN uses API workers, this change also introduces the
callbacks necessary in pf_plugin, so events related to port
forwarding are sent using neutron_lib callbacks registry.

Related-Bug: #1877447
Change-Id: I8124fac13bf4d802d232e8b3976e6a2cebc72106

devstack/plugin.sh

@@ -110,6 +110,9 @@ if [[ "$1" == "stack" ]]; then
                 configure_ml2_extension_drivers
             fi
             if is_ovn_enabled; then
+                if is_service_enabled q-port-forwarding neutron-port-forwarding; then
+                    configure_port_forwarding
+                fi
                 configure_ovn_plugin
                 start_ovn
             fi

neutron/services/portforwarding/callbacks.py

@@ -0,0 +1,28 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+
+# TODO(flaviof): remove this once moved over to neutron-lib payloads
+class PortForwardingPayload(object):
+    """Payload for port-forwarding-related callback registry notifications."""
+
+    def __init__(self, context, current_pf=None, original_pf=None):
+        self.context = context
+        self.current_pf = current_pf
+        self.original_pf = original_pf
+
+    def __eq__(self, other):
+        return (isinstance(other, self.__class__) and
+                self.__dict__ == other.__dict__)
+
+    def __ne__(self, other):
+        return not self.__eq__(other)

neutron/services/portforwarding/constants.py

@@ -0,0 +1,19 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+# TODO(flaviof): This file is a place holder. Everything here should move to
+#                neutron-lib someday.
+
+# String literals representing core resources.
+# TODO(flaviof): move to neutron_lib/callbacks/resources.py
+PORT_FORWARDING = 'port_forwarding'
+PORT_FORWARDING_PLUGIN = 'port_forwarding_plugin'

neutron/services/portforwarding/pf_plugin.py

@@ -14,6 +14,7 @@
 #    under the License.
 
 import collections
+import copy
 
 import netaddr
 from neutron_lib.api.definitions import expose_port_forwarding_in_fip
@@ -31,6 +32,7 @@ from neutron_lib.exceptions import l3 as lib_l3_exc
 from neutron_lib.objects import exceptions as obj_exc
 from neutron_lib.plugins import constants
 from neutron_lib.plugins import directory
+from oslo_config import cfg
 from oslo_log import log as logging
 
 from neutron._i18n import _
@@ -43,7 +45,9 @@ from neutron.extensions import floating_ip_port_forwarding as fip_pf
 from neutron.objects import base as base_obj
 from neutron.objects import port_forwarding as pf
 from neutron.objects import router as l3_obj
+from neutron.services.portforwarding import callbacks
 from neutron.services.portforwarding.common import exceptions as pf_exc
+from neutron.services.portforwarding import constants as pf_consts
 
 LOG = logging.getLogger(__name__)
 
@@ -51,6 +55,18 @@ LOG = logging.getLogger(__name__)
 PORT_FORWARDING_FLOATINGIP_KEY = '_pf_floatingips'
 
 
+def _required_service_plugins():
+    supported_svc_plugins = [l3.ROUTER, 'ovn-router']
+    try:
+        plugins = [svc for svc in supported_svc_plugins if
+                   svc in cfg.CONF.service_plugins]
+    except cfg.NoSuchOptError:
+        plugins = None
+        pass
+    # Use l3.ROUTER as required service plugin if no other was provided.
+    return plugins or [l3.ROUTER]
+
+
 @resource_extend.has_resource_extenders
 @registry.has_registry_receivers
 class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
@@ -59,7 +75,7 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
     This class implements a Port Forwarding plugin.
     """
 
-    required_service_plugins = ['router']
+    required_service_plugins = _required_service_plugins()
 
     supported_extension_aliases = [apidef.ALIAS,
                                    expose_port_forwarding_in_fip.ALIAS,
@@ -74,6 +90,8 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
         self.push_api = resources_rpc.ResourcesPushRpcApi()
         self.l3_plugin = directory.get_plugin(constants.L3)
         self.core_plugin = directory.get_plugin()
+        registry.publish(pf_consts.PORT_FORWARDING_PLUGIN, events.AFTER_INIT,
+                         self)
 
     @staticmethod
     @resource_extend.extends([l3.FLOATINGIPS])
@@ -225,6 +243,11 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
 
         self.push_api.push(context, remove_port_forwarding_list,
                            rpc_events.DELETED)
+        registry_notify_payload = [
+            callbacks.PortForwardingPayload(context, original_pf=pf_obj) for
+            pf_obj in remove_port_forwarding_list]
+        registry.notify(pf_consts.PORT_FORWARDING, events.AFTER_DELETE, self,
+                        payload=registry_notify_payload)
 
     def _get_internal_ip_subnet(self, request_ip, fixed_ips):
         request_ip = netaddr.IPNetwork(request_ip)
@@ -347,6 +370,10 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
                                          msg=message)
 
             self.push_api.push(context, [pf_obj], rpc_events.CREATED)
+            registry.notify(pf_consts.PORT_FORWARDING, events.AFTER_CREATE,
+                            self,
+                            payload=[callbacks.PortForwardingPayload(context,
+                                current_pf=pf_obj)])
             return pf_obj
 
     @db_base_plugin_common.convert_result_to_dict
@@ -364,6 +391,7 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
                 pf_obj = pf.PortForwarding.get_object(context, id=id)
                 if not pf_obj:
                     raise pf_exc.PortForwardingNotFound(id=id)
+                original_pf_obj = copy.deepcopy(pf_obj)
                 ori_internal_port_id = pf_obj.internal_port_id
                 if new_internal_port_id and (new_internal_port_id !=
                                              ori_internal_port_id):
@@ -396,6 +424,9 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
             raise lib_exc.BadRequest(resource=apidef.RESOURCE_NAME,
                                      msg=message)
         self.push_api.push(context, [pf_obj], rpc_events.UPDATED)
+        registry.notify(pf_consts.PORT_FORWARDING, events.AFTER_UPDATE, self,
+                        payload=[callbacks.PortForwardingPayload(context,
+                            current_pf=pf_obj, original_pf=original_pf_obj)])
         return pf_obj
 
     def _check_router_match(self, context, fip_obj, router_id, pf_dict):
@@ -494,6 +525,9 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
                 fip_obj.update()
             pf_obj.delete()
         self.push_api.push(context, [pf_obj], rpc_events.DELETED)
+        registry.notify(pf_consts.PORT_FORWARDING, events.AFTER_DELETE, self,
+                        payload=[callbacks.PortForwardingPayload(
+                            context, original_pf=pf_obj)])
 
     def sync_port_forwarding_fip(self, context, routers):
         if not routers:

neutron/tests/unit/services/portforwarding/test_pf_plugin.py

@@ -13,8 +13,11 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+from collections import namedtuple
 from unittest import mock
 
+from neutron_lib.callbacks import events
+from neutron_lib.callbacks import registry
 from neutron_lib import context
 from neutron_lib import exceptions as lib_exc
 from neutron_lib.exceptions import l3 as lib_l3_exc
@@ -35,6 +38,7 @@ from neutron.objects import base as obj_base
 from neutron.objects import port_forwarding
 from neutron.objects import router
 from neutron.services.portforwarding.common import exceptions as pf_exc
+from neutron.services.portforwarding import constants as pf_consts
 from neutron.services.portforwarding import pf_plugin
 from neutron.tests.unit import testlib_api
 
@@ -63,6 +67,7 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
             prod_registry, '_get_manager', return_value=self.prod_mgr).start()
         self.setup_coreplugin(load_plugins=False)
 
+        mock.patch('neutron_lib.callbacks.registry.publish').start()
         mock.patch('neutron.objects.db.api.create_object').start()
         mock.patch('neutron.objects.db.api.update_object').start()
         mock.patch('neutron.objects.db.api.delete_object').start()
@@ -102,13 +107,14 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         get_objects_mock.assert_called_once_with(
             self.ctxt, _pager=mock.ANY, floatingip_id=None)
 
+    @mock.patch.object(registry, 'notify')
     @mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
     @mock.patch.object(port_forwarding.PortForwarding, 'get_object')
     @mock.patch.object(port_forwarding.PortForwarding, 'get_objects')
     @mock.patch.object(router.FloatingIP, 'get_object')
     def test_delete_floatingip_port_forwarding(
             self, fip_get_object_mock, pf_get_objects_mock,
-            pf_get_object_mock, push_api_mock):
+            pf_get_object_mock, push_api_mock, registry_notify_mock):
 
         # After delete, not empty resource list
         pf_get_objects_mock.return_value = [mock.Mock(id='pf_id'),
@@ -122,11 +128,15 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         pf_obj.delete.assert_called()
         push_api_mock.assert_called_once_with(
             self.ctxt, mock.ANY, rpc_events.DELETED)
+        registry_notify_mock.assert_called_once_with(
+            pf_consts.PORT_FORWARDING,
+            events.AFTER_DELETE, self.pf_plugin, payload=mock.ANY)
 
         # After delete, empty resource list
         pf_get_objects_mock.reset_mock()
         pf_get_object_mock.reset_mock()
         push_api_mock.reset_mock()
+        registry_notify_mock.reset_mock()
         pf_obj = mock.Mock(id='need_to_delete_pf_id', floatingip_id='fip_id')
         fip_obj = mock.Mock(id='fip_id')
         fip_get_object_mock.return_value = fip_obj
@@ -144,6 +154,9 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         fip_obj.update.assert_called()
         push_api_mock.assert_called_once_with(
             self.ctxt, mock.ANY, rpc_events.DELETED)
+        registry_notify_mock.assert_called_once_with(
+            pf_consts.PORT_FORWARDING,
+            events.AFTER_DELETE, self.pf_plugin, payload=mock.ANY)
 
     @mock.patch.object(port_forwarding.PortForwarding, 'get_object')
     def test_negative_delete_floatingip_port_forwarding(
@@ -154,10 +167,11 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
             self.pf_plugin.delete_floatingip_port_forwarding,
             self.ctxt, 'pf_id', floatingip_id='fip_id')
 
+    @mock.patch.object(registry, 'notify')
     @mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
     @mock.patch.object(port_forwarding.PortForwarding, 'get_object')
     def test_update_floatingip_port_forwarding(
-            self, mock_pf_get_object, mock_rpc_push):
+            self, mock_pf_get_object, mock_rpc_push, mock_registry_notify):
         pf_input = {
             'port_forwarding':
                 {'port_forwarding': {
@@ -173,6 +187,9 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         self.assertTrue(pf_obj.update)
         mock_rpc_push.assert_called_once_with(
             self.ctxt, mock.ANY, rpc_events.UPDATED)
+        mock_registry_notify.assert_called_once_with(
+            pf_consts.PORT_FORWARDING,
+            events.AFTER_UPDATE, self.pf_plugin, payload=mock.ANY)
 
     @mock.patch.object(port_forwarding.PortForwarding, 'get_object')
     def test_negative_update_floatingip_port_forwarding(
@@ -192,6 +209,7 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
     @mock.patch.object(pf_plugin.PortForwardingPlugin,
                        '_check_port_has_binding_floating_ip')
     @mock.patch.object(obj_base.NeutronDbObject, 'update_objects')
+    @mock.patch.object(registry, 'notify')
     @mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
     @mock.patch.object(pf_plugin.PortForwardingPlugin, '_check_router_match')
     @mock.patch.object(pf_plugin.PortForwardingPlugin,
@@ -201,8 +219,8 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
     @mock.patch('neutron.objects.port_forwarding.PortForwarding')
     def test_create_floatingip_port_forwarding(
             self, mock_port_forwarding, mock_fip_get_object, mock_find_router,
-            mock_check_router_match, mock_push_api, mock_update_objects,
+            mock_check_router_match, mock_push_api, mock_registry_notify,
-            mock_check_bind_fip):
+            mock_update_objects, mock_check_bind_fip):
         # Update fip
         pf_input = {
             'port_forwarding':
@@ -224,6 +242,9 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         self.assertTrue(pf_obj.create.called)
         mock_push_api.assert_called_once_with(
             self.ctxt, mock.ANY, rpc_events.CREATED)
+        mock_registry_notify.assert_called_once_with(
+            pf_consts.PORT_FORWARDING,
+            events.AFTER_CREATE, self.pf_plugin, payload=mock.ANY)
 
         # Not update fip
         pf_obj.reset_mock()
@@ -231,6 +252,7 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         mock_port_forwarding.reset_mock()
         mock_update_objects.reset_mock()
         mock_push_api.reset_mock()
+        mock_registry_notify.reset_mock()
         mock_port_forwarding.return_value = pf_obj
         fip_obj.router_id = 'router_id'
         fip_obj.fixed_port_id = ''
@@ -242,6 +264,9 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         self.assertFalse(mock_update_objects.called)
         mock_push_api.assert_called_once_with(
             self.ctxt, mock.ANY, rpc_events.CREATED)
+        mock_registry_notify.assert_called_once_with(
+            pf_consts.PORT_FORWARDING,
+            events.AFTER_CREATE, self.pf_plugin, payload=mock.ANY)
 
     @mock.patch.object(pf_plugin.PortForwardingPlugin,
                        '_check_port_has_binding_floating_ip')
@@ -367,3 +392,39 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
         self.assertRaises(pf_exc.PortHasBindingFloatingIP,
                           self.pf_plugin.update_floatingip_port_forwarding,
                           self.ctxt, 'fake-pf-id', 'fip_id_2', **pf_input)
+
+    def test_service_plugins_values(self):
+        exp_default = ['router']
+        supported_plugins = ['router', 'ovn-router']
+        same_as_input = 'same_as_input'
+        TC = namedtuple('TC', 'input expected description')
+        test_cases = [
+            TC([], exp_default, "default from empty cfg"),
+            TC(['foo'], exp_default, "default from unexpected cfg"),
+            TC(['foo', 123], exp_default, "default from unexpected cfg"),
+            TC(['foo', 'router'], exp_default, "default from valid cfg"),
+            TC(['router'], same_as_input, "valid cfg 1"),
+            TC(['router'], same_as_input, "valid cfg 1"),
+            TC(['ovn-router'], same_as_input, "valid cfg 2"),
+            TC(['ovn-router', 'router'], supported_plugins, "valid cfg 3"),
+            TC(['router', 'ovn-router'], supported_plugins, "valid cfg 4"),
+            TC(['bar', 'router', 'foo'], ['router'], "valid cfg 5"),
+            TC(['bar', 'ovn-router', 'foo'], ['ovn-router'], "valid cfg 6"),
+            TC(['bar', 'router', 123, 'ovn-router', 'foo', 'kitchen', 'sink'],
+               supported_plugins, "valid cfg 7"),
+        ]
+        for tc in test_cases:
+            cfg.CONF.set_override("service_plugins", tc.input)
+            result = pf_plugin._required_service_plugins()
+            if tc.expected == same_as_input:
+                self.assertEqual(tc.input, result, tc.description)
+            else:
+                self.assertEqual(tc.expected, result, tc.description)
+
+    @mock.patch.object(cfg.ConfigOpts, '__getattr__')
+    def test_service_plugins_no_such_opt(self, mock_config_opts_get):
+        description = "test cfg.NoSuchOptError exception"
+        mock_config_opts_get.side_effect = cfg.NoSuchOptError('test_svc_plug')
+        result = pf_plugin._required_service_plugins()
+        mock_config_opts_get.assert_called_once()
+        self.assertEqual(['router'], result, description)