NSXV+NSXV3: add support for pluggable extensions

A new configuration variable nsx_extension_drivers
has been added. This is in the DEFAULT section. This enable us
to code support to add via configurations extensions, for
example dns_integration.

Co-authored-by: Shih-Hao Li <shihli@vmware.com>

Change-Id: Iea4715522d9c7cf327b7f1a751b78f14d5e06e75

releasenotes/notes/nsx-extension-drivers-b1aedabe5296d4d0.yaml

@@ -0,0 +1,7 @@
+---
+prelude: >
+    We have added a new configuration variable that will enable us to
+    enable existing extensions. The new configuration variable is
+    ``nsx_extension_drivers``. This is in the default section.
+    This is a list of extansion names. The code for the drivers
+    must be in the directory vmware_nsx.extension_drivers.

vmware_nsx/common/config.py

@@ -245,6 +245,11 @@ nsx_common_opts = [
                        "specify the id of resources. This should only "
                        "be enabled in order to allow one to migrate an "
                        "existing install of neutron to the nsx-v3 plugin.")),
+    cfg.ListOpt('nsx_extension_drivers',
+                default=[],
+                help=_("An ordered list of extension driver "
+                       "entrypoints to be loaded from the "
+                       "vmware_nsx.extension_drivers namespace.")),
 ]
 
 nsx_v3_opts = [

vmware_nsx/common/driver_api.py

@@ -0,0 +1,157 @@
+# Copyright (c) 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+#    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.
+
+import abc
+
+import six
+
+
+@six.add_metaclass(abc.ABCMeta)
+class ExtensionDriver(object):
+    """Define stable abstract interface for extension drivers.
+    An extension driver extends the core resources implemented by the
+    plugin with additional attributes. Methods that process create
+    and update operations for these resources validate and persist
+    values for extended attributes supplied through the API. Other
+    methods extend the resource dictionaries returned from the API
+    operations with the values of the extended attributes.
+    """
+
+    @abc.abstractmethod
+    def initialize(self):
+        """Perform driver initialization.
+        Called after all drivers have been loaded and the database has
+        been initialized. No abstract methods defined below will be
+        called prior to this method being called.
+        """
+        pass
+
+    @property
+    def extension_alias(self):
+        """Supported extension alias.
+        Return the alias identifying the core API extension supported
+        by this driver. Do not declare if API extension handling will
+        be left to a service plugin, and we just need to provide
+        core resource extension and updates.
+        """
+        pass
+
+    def process_create_network(self, plugin_context, data, result):
+        """Process extended attributes for create network.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming network data
+        :param result: network dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and persist any extended network attributes defined by this
+        driver. Extended attribute values must also be added to
+        result.
+        """
+        pass
+
+    def process_create_subnet(self, plugin_context, data, result):
+        """Process extended attributes for create subnet.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming subnet data
+        :param result: subnet dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and persist any extended subnet attributes defined by this
+        driver. Extended attribute values must also be added to
+        result.
+        """
+        pass
+
+    def process_create_port(self, plugin_context, data, result):
+        """Process extended attributes for create port.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming port data
+        :param result: port dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and persist any extended port attributes defined by this
+        driver. Extended attribute values must also be added to
+        result.
+        """
+        pass
+
+    def process_update_network(self, plugin_context, data, result):
+        """Process extended attributes for update network.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming network data
+        :param result: network dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and update any extended network attributes defined by this
+        driver. Extended attribute values, whether updated or not,
+        must also be added to result.
+        """
+        pass
+
+    def process_update_subnet(self, plugin_context, data, result):
+        """Process extended attributes for update subnet.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming subnet data
+        :param result: subnet dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and update any extended subnet attributes defined by this
+        driver. Extended attribute values, whether updated or not,
+        must also be added to result.
+        """
+        pass
+
+    def process_update_port(self, plugin_context, data, result):
+        """Process extended attributes for update port.
+        :param plugin_context: plugin request context
+        :param data: dictionary of incoming port data
+        :param result: port dictionary to extend
+        Called inside transaction context on plugin_context.session to
+        validate and update any extended port attributes defined by this
+        driver. Extended attribute values, whether updated or not,
+        must also be added to result.
+        """
+        pass
+
+    def extend_network_dict(self, session, base_model, result):
+        """Add extended attributes to network dictionary.
+        :param session: database session
+        :param base_model: network model data
+        :param result: network dictionary to extend
+        Called inside transaction context on session to add any
+        extended attributes defined by this driver to a network
+        dictionary to be used for driver calls and/or
+        returned as the result of a network operation.
+        """
+        pass
+
+    def extend_subnet_dict(self, session, base_model, result):
+        """Add extended attributes to subnet dictionary.
+        :param session: database session
+        :param base_model: subnet model data
+        :param result: subnet dictionary to extend
+        Called inside transaction context on session to add any
+        extended attributes defined by this driver to a subnet
+        dictionary to be used for driver calls and/or
+        returned as the result of a subnet operation.
+        """
+        pass
+
+    def extend_port_dict(self, session, base_model, result):
+        """Add extended attributes to port dictionary.
+        :param session: database session
+        :param base_model: port model data
+        :param result: port dictionary to extend
+        Called inside transaction context on session to add any
+        extended attributes defined by this driver to a port
+        dictionary to be used for driver calls
+        and/or returned as the result of a port operation.
+        """
+        pass

vmware_nsx/common/managers.py

@@ -0,0 +1,134 @@
+# Copyright (c) 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+#    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.
+
+from oslo_config import cfg
+from oslo_log import log
+from oslo_utils import excutils
+import stevedore
+
+from vmware_nsx._i18n import _LE, _LI
+
+LOG = log.getLogger(__name__)
+
+
+class ExtensionManager(stevedore.named.NamedExtensionManager):
+    """Manage extension drivers using drivers."""
+
+    def __init__(self):
+        # Ordered list of extension drivers, defining
+        # the order in which the drivers are called.
+        self.ordered_ext_drivers = []
+
+        LOG.info(_LI("Configured extension driver names: %s"),
+                 cfg.CONF.nsx_extension_drivers)
+        super(ExtensionManager, self).__init__('vmware_nsx.extension_drivers',
+                                               cfg.CONF.nsx_extension_drivers,
+                                               invoke_on_load=True,
+                                               name_order=True)
+        LOG.info(_LI("Loaded extension driver names: %s"), self.names())
+        self._register_drivers()
+
+    def _register_drivers(self):
+        """Register all extension drivers.
+
+        This method should only be called once in the ExtensionManager
+        constructor.
+        """
+        for ext in self:
+            self.ordered_ext_drivers.append(ext)
+        LOG.info(_LI("Registered extension drivers: %s"),
+                 [driver.name for driver in self.ordered_ext_drivers])
+
+    def initialize(self):
+        # Initialize each driver in the list.
+        for driver in self.ordered_ext_drivers:
+            LOG.info(_LI("Initializing extension driver '%s'"), driver.name)
+            driver.obj.initialize()
+
+    def extension_aliases(self):
+        exts = []
+        for driver in self.ordered_ext_drivers:
+            alias = driver.obj.extension_alias
+            if alias:
+                exts.append(alias)
+                LOG.info(_LI("Got %(alias)s extension from driver '%(drv)s'"),
+                         {'alias': alias, 'drv': driver.name})
+        return exts
+
+    def _call_on_ext_drivers(self, method_name, plugin_context, data, result):
+        """Helper method for calling a method across all extension drivers."""
+        for driver in self.ordered_ext_drivers:
+            try:
+                getattr(driver.obj, method_name)(plugin_context, data, result)
+            except Exception:
+                with excutils.save_and_reraise_exception():
+                    LOG.info(_LI("Extension driver '%(name)s' failed in "
+                             "%(method)s"),
+                             {'name': driver.name, 'method': method_name})
+
+    def process_create_network(self, plugin_context, data, result):
+        """Notify all extension drivers during network creation."""
+        self._call_on_ext_drivers("process_create_network", plugin_context,
+                                  data, result)
+
+    def process_update_network(self, plugin_context, data, result):
+        """Notify all extension drivers during network update."""
+        self._call_on_ext_drivers("process_update_network", plugin_context,
+                                  data, result)
+
+    def process_create_subnet(self, plugin_context, data, result):
+        """Notify all extension drivers during subnet creation."""
+        self._call_on_ext_drivers("process_create_subnet", plugin_context,
+                                  data, result)
+
+    def process_update_subnet(self, plugin_context, data, result):
+        """Notify all extension drivers during subnet update."""
+        self._call_on_ext_drivers("process_update_subnet", plugin_context,
+                                  data, result)
+
+    def process_create_port(self, plugin_context, data, result):
+        """Notify all extension drivers during port creation."""
+        self._call_on_ext_drivers("process_create_port", plugin_context,
+                                  data, result)
+
+    def process_update_port(self, plugin_context, data, result):
+        """Notify all extension drivers during port update."""
+        self._call_on_ext_drivers("process_update_port", plugin_context,
+                                  data, result)
+
+    def _call_on_dict_driver(self, method_name, session, base_model, result):
+        for driver in self.ordered_ext_drivers:
+            try:
+                getattr(driver.obj, method_name)(session, base_model, result)
+            except Exception:
+                LOG.error(_LE("Extension driver '%(name)s' failed in "
+                          "%(method)s"),
+                          {'name': driver.name, 'method': method_name})
+                raise
+
+    def extend_network_dict(self, session, base_model, result):
+        """Notify all extension drivers to extend network dictionary."""
+        self._call_on_dict_driver("extend_network_dict", session, base_model,
+                                  result)
+
+    def extend_subnet_dict(self, session, base_model, result):
+        """Notify all extension drivers to extend subnet dictionary."""
+        self._call_on_dict_driver("extend_subnet_dict", session, base_model,
+                                  result)
+
+    def extend_port_dict(self, session, base_model, result):
+        """Notify all extension drivers to extend port dictionary."""
+        self._call_on_dict_driver("extend_port_dict", session, base_model,
+                                  result)

vmware_nsx/extension_drivers/__init__.py

@@ -0,0 +1,16 @@
+#  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.
+
+
+import os
+
+NSX_EXT_PATH = os.path.join(os.path.dirname(__file__), 'extensions')

vmware_nsx/plugins/nsx_v/plugin.py

@@ -85,6 +85,7 @@ from vmware_nsx._i18n import _, _LE, _LI, _LW
 from vmware_nsx.common import config  # noqa
 from vmware_nsx.common import exceptions as nsx_exc
 from vmware_nsx.common import locking
+from vmware_nsx.common import managers as nsx_managers
 from vmware_nsx.common import nsx_constants
 from vmware_nsx.common import nsxv_constants
 from vmware_nsx.common import utils as c_utils
@@ -186,11 +187,15 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         router=l3_db_models.Router,
         floatingip=l3_db_models.FloatingIP)
     def __init__(self):
+        self._extension_manager = nsx_managers.ExtensionManager()
         super(NsxVPluginV2, self).__init__()
         self.init_is_complete = False
         registry.subscribe(self.init_complete,
                            resources.PROCESS,
                            events.AFTER_INIT)
+        self._extension_manager.initialize()
+        self.supported_extension_aliases.extend(
+            self._extension_manager.extension_aliases())
         self.metadata_proxy_handler = None
         config.validate_nsxv_config_options()
         neutron_extensions.append_api_extensions_path(
@@ -266,6 +271,16 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         if c_utils.is_nsxv_version_6_2(self.nsx_v.vcns.get_version()):
             self.supported_extension_aliases.append("provider-security-group")
 
+    # Register extend dict methods for network and port resources.
+    # Each extension driver that supports extend attribute for the resources
+    # can add those attribute to the result.
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attr.NETWORKS, ['_ext_extend_network_dict'])
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attr.PORTS, ['_ext_extend_port_dict'])
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attr.SUBNETS, ['_ext_extend_subnet_dict'])
+
     def init_complete(self, resource, event, trigger, **kwargs):
         has_metadata_cfg = (
             cfg.CONF.nsxv.nova_metadata_ips
@@ -334,6 +349,22 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         self.start_rpc_listeners_called = True
         return self.conn.consume_in_threads()
 
+    def _ext_extend_network_dict(self, result, netdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_network_dict(session, netdb, result)
+
+    def _ext_extend_port_dict(self, result, portdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_port_dict(session, portdb, result)
+
+    def _ext_extend_subnet_dict(self, result, subnetdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_subnet_dict(
+                session, subnetdb, result)
+
     def _create_security_group_container(self):
         name = "OpenStack Security Group container"
         with locking.LockManager.get_lock('security-group-container-init'):
@@ -1006,6 +1037,8 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
             with context.session.begin(subtransactions=True):
                 new_net = super(NsxVPluginV2, self).create_network(context,
                                                                    network)
+                self._extension_manager.process_create_network(
+                    context, net_data, new_net)
                 # Process port security extension
                 self._process_network_port_security_create(
                     context, net_data, new_net)
@@ -1283,6 +1316,8 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         with context.session.begin(subtransactions=True):
             net_res = super(NsxVPluginV2, self).update_network(context, id,
                                                                network)
+            self._extension_manager.process_update_network(context, net_attrs,
+                                                           net_res)
             self._process_network_port_security_update(
                 context, net_attrs, net_res)
             self._process_l3_update(context, net_res, net_attrs)
@@ -1342,13 +1377,18 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
 
     @db_api.retry_db_errors
     def base_create_port(self, context, port):
-        return super(NsxVPluginV2, self).create_port(context, port)
+        created_port = super(NsxVPluginV2, self).create_port(context, port)
+        self._extension_manager.process_create_port(
+            context, port['port'], created_port)
+        return created_port
 
     def create_port(self, context, port):
         port_data = port['port']
         with context.session.begin(subtransactions=True):
             # First we allocate port in neutron database
             neutron_db = super(NsxVPluginV2, self).create_port(context, port)
+            self._extension_manager.process_create_port(
+                context, port_data, neutron_db)
             # Port port-security is decided by the port-security state on the
             # network it belongs to, unless specifically specified here
             if validators.is_attr_set(port_data.get(psec.PORTSECURITY)):
@@ -1591,6 +1631,8 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         with context.session.begin(subtransactions=True):
             ret_port = super(NsxVPluginV2, self).update_port(
                 context, id, port)
+            self._extension_manager.process_update_port(
+                context, port_data, ret_port)
             # copy values over - except fixed_ips as
             # they've already been processed
             updates_fixed_ips = port['port'].pop('fixed_ips', [])
@@ -1994,6 +2036,8 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         with locking.LockManager.get_lock(subnet['subnet']['network_id']):
             with locking.LockManager.get_lock('nsx-edge-pool'):
                 s = super(NsxVPluginV2, self).create_subnet(context, subnet)
+            self._extension_manager.process_create_subnet(
+                context, subnet['subnet'], s)
             if s['enable_dhcp']:
                 try:
                     self._process_subnet_ext_attr_create(
@@ -2075,6 +2119,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
                                          orig_enable_dhcp=enable_dhcp,
                                          orig_host_routes=orig_host_routes)
         subnet = super(NsxVPluginV2, self).update_subnet(context, id, subnet)
+        self._extension_manager.process_update_subnet(context, s, subnet)
         update_dhcp_config = self._process_subnet_ext_attr_update(
             context.session, subnet, s)
         if (gateway_ip != subnet['gateway_ip'] or update_dhcp_config or
@@ -2715,9 +2760,11 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
         since the actual backend work was already done by the router driver,
         and it may cause a deadlock.
         """
-        super(NsxVPluginV2, self).update_port(context,
+        port_data = {'fixed_ips': ext_ips}
-                                              router.gw_port['id'],
+        updated_port = super(NsxVPluginV2, self).update_port(
-                                              {'port': {'fixed_ips': ext_ips}})
+            context, router.gw_port['id'], {'port': port_data})
+        self._extension_manager.process_update_port(
+            context, port_data, updated_port)
         context.session.expire(router.gw_port)
 
     def _update_router_gw_info(self, context, router_id, info,

vmware_nsx/plugins/nsx_v3/plugin.py

@@ -22,6 +22,7 @@ from neutron.api.rpc.callbacks import resources as callbacks_resources
 from neutron.api.rpc.handlers import dhcp_rpc
 from neutron.api.rpc.handlers import metadata_rpc
 from neutron.api.rpc.handlers import resources_rpc
+from neutron.api.v2 import attributes
 from neutron.callbacks import events
 from neutron.callbacks import exceptions as callback_exc
 from neutron.callbacks import registry
@@ -33,6 +34,7 @@ from neutron.db import _utils as db_utils
 from neutron.db import agents_db
 from neutron.db import agentschedulers_db
 from neutron.db import allowedaddresspairs_db as addr_pair_db
+from neutron.db import api as db_api
 from neutron.db import db_base_plugin_v2
 from neutron.db import dns_db
 from neutron.db import external_net_db
@@ -77,6 +79,7 @@ from vmware_nsx.api_replay import utils as api_replay_utils
 from vmware_nsx.common import config  # noqa
 from vmware_nsx.common import exceptions as nsx_exc
 from vmware_nsx.common import locking
+from vmware_nsx.common import managers
 from vmware_nsx.common import utils
 from vmware_nsx.db import db as nsx_db
 from vmware_nsx.db import extended_security_group
@@ -166,9 +169,12 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
         router=l3_db_models.Router,
         floatingip=l3_db_models.FloatingIP)
     def __init__(self):
+        self._extension_manager = managers.ExtensionManager()
         super(NsxV3Plugin, self).__init__()
         LOG.info(_LI("Starting NsxV3Plugin"))
-
+        self._extension_manager.initialize()
+        self.supported_extension_aliases.extend(
+            self._extension_manager.extension_aliases())
         self.nsxlib = v3_utils.get_nsxlib_wrapper()
         # reinitialize the cluster upon fork for api workers to ensure each
         # process has its own keepalive loops + state
@@ -224,6 +230,16 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
         # Register NSXv3 trunk driver to support trunk extensions
         self.trunk_driver = trunk_driver.NsxV3TrunkDriver.create(self)
 
+    # Register extend dict methods for network and port resources.
+    # Each extension driver that supports extend attribute for the resources
+    # can add those attribute to the result.
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attributes.NETWORKS, ['_ext_extend_network_dict'])
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attributes.PORTS, ['_ext_extend_port_dict'])
+    db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
+        attributes.SUBNETS, ['_ext_extend_subnet_dict'])
+
     def _init_nsx_profiles(self):
         LOG.debug("Initializing NSX v3 port spoofguard switching profile")
         if not self._init_port_security_profile():
@@ -538,6 +554,22 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
 
         return self.conn.consume_in_threads()
 
+    def _ext_extend_network_dict(self, result, netdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_network_dict(session, netdb, result)
+
+    def _ext_extend_port_dict(self, result, portdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_port_dict(session, portdb, result)
+
+    def _ext_extend_subnet_dict(self, result, subnetdb):
+        session = db_api.get_session()
+        with session.begin(subtransactions=True):
+            self._extension_manager.extend_subnet_dict(
+                session, subnetdb, result)
+
     def _validate_provider_create(self, context, network_data):
         is_provider_net = any(
             validators.is_attr_set(network_data.get(f))
@@ -726,7 +758,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                 # Create network in Neutron
                 created_net = super(NsxV3Plugin, self).create_network(context,
                                                                       network)
-
+                self._extension_manager.process_create_network(
+                    context, net_data, created_net)
                 if psec.PORTSECURITY not in net_data:
                     net_data[psec.PORTSECURITY] = True
                 self._process_network_port_security_create(
@@ -891,7 +924,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
             self._assert_on_external_net_with_qos(net_data)
         updated_net = super(NsxV3Plugin, self).update_network(context, id,
                                                               network)
-
+        self._extension_manager.process_update_network(context, net_data,
+                                                       updated_net)
         if psec.PORTSECURITY in network['network']:
             self._process_network_port_security_update(
                 context, network['network'], updated_net)
@@ -1201,6 +1235,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                     if self._has_no_dhcp_enabled_subnet(context, network):
                         created_subnet = super(
                             NsxV3Plugin, self).create_subnet(context, subnet)
+                        self._extension_manager.process_create_subnet(context,
+                            subnet['subnet'], created_subnet)
                         self._enable_native_dhcp(context, network,
                                                  created_subnet)
                         msg = None
@@ -1259,6 +1295,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                                 updated_subnet = super(
                                     NsxV3Plugin, self).update_subnet(
                                     context, subnet_id, subnet)
+                                self._extension_manager.process_update_subnet(
+                                    context, subnet['subnet'], updated_subnet)
                                 self._enable_native_dhcp(context, network,
                                                          updated_subnet)
                                 msg = None
@@ -1279,10 +1317,14 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                         updated_subnet = super(
                             NsxV3Plugin, self).update_subnet(
                             context, subnet_id, subnet)
+                        self._extension_manager.process_update_subnet(
+                            context, subnet['subnet'], updated_subnet)
 
         if not updated_subnet:
             updated_subnet = super(NsxV3Plugin, self).update_subnet(
                 context, subnet_id, subnet)
+            self._extension_manager.process_update_subnet(
+                context, subnet['subnet'], updated_subnet)
 
         # Check if needs to update logical DHCP server for native DHCP.
         if (cfg.CONF.nsx_v3.native_dhcp_metadata and
@@ -1881,6 +1923,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
                 self._assert_on_external_net_port_with_qos(port_data)
 
             neutron_db = super(NsxV3Plugin, self).create_port(context, port)
+            self._extension_manager.process_create_port(
+                context, port_data, neutron_db)
             port["port"].update(neutron_db)
 
             (is_psec_on, has_ip) = self._create_port_preprocess_security(
@@ -2242,7 +2286,8 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
             old_mac_learning_state = original_port.get(mac_ext.MAC_LEARNING)
             updated_port = super(NsxV3Plugin, self).update_port(context,
                                                                 id, port)
-
+            self._extension_manager.process_update_port(context, port['port'],
+                                                        updated_port)
             # copy values over - except fixed_ips as
             # they've already been processed
             port['port'].pop('fixed_ips', None)