Move notify_security_groups_member_updated to callback

This moves the notify_security_groups_member_updated calls
outside of ML2 into a callback in the security groups RPC
base.

This will allow us to ultimately stop overriding the
create/delete/update port inheritance in ML2.

This results in a small behavior change internally for bulk
port creation operations where the notifier will be called
once for each port rather than once for all ports. We can
revisit this in the future if it turns out to be problematic
but right now the bulk port creation API is rarely anyway
since it's mostly a hidden feature (lack of client support).

Partial-Bug: #1666424
Change-Id: I344a82d5a1a8bf1e2b3f6a3ce7725682bb03b007

neutron/db/securitygroups_rpc_base.py

@@ -19,6 +19,9 @@ from neutron_lib.utils import helpers
 from oslo_log import log as logging
 
 from neutron._i18n import _
+from neutron.callbacks import events
+from neutron.callbacks import registry
+from neutron.callbacks import resources
 from neutron.db import api as db_api
 from neutron.db.models import allowed_address_pair as aap_models
 from neutron.db.models import securitygroup as sg_models
@@ -35,9 +38,18 @@ DIRECTION_IP_PREFIX = {'ingress': 'source_ip_prefix',
 DHCP_RULE_PORT = {4: (67, 68, const.IPv4), 6: (547, 546, const.IPv6)}
 
 
+@registry.has_registry_receivers
 class SecurityGroupServerRpcMixin(sg_db.SecurityGroupDbMixin):
     """Mixin class to add agent-based security group implementation."""
 
+    @registry.receives(resources.PORT, [events.AFTER_CREATE,
+                                        events.AFTER_UPDATE,
+                                        events.AFTER_DELETE])
+    def notify_sg_on_port_change(self, resource, event, trigger, context,
+                                 port, *args, **kwargs):
+        """Trigger notification to other SG members on port changes."""
+        self.notify_security_groups_member_updated(context, port)
+
     def get_port_from_device(self, context, device):
         """Get port dict from device name on an agent.
 

neutron/plugins/ml2/plugin.py

@@ -1090,11 +1090,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
                 LOG.error(_LE("mechanism_manager.create_port_postcommit "
                               "failed, deleting port '%s'"), result['id'])
                 self.delete_port(context, result['id'], l3_port_check=False)
-
-        # REVISIT(rkukura): Is there any point in calling this before
-        # a binding has been successfully established?
-        self.notify_security_groups_member_updated(context, result)
-
         try:
             bound_context = self._bind_port_if_needed(mech_context)
         except ml2_exc.MechanismDriverError:
@@ -1110,11 +1105,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
     def create_port_bulk(self, context, ports):
         objects = self._create_bulk_ml2(attributes.PORT, context, ports)
 
-        # REVISIT(rkukura): Is there any point in calling this before
-        # a binding has been successfully established?
-        results = [obj['result'] for obj in objects]
-        self.notify_security_groups_member_updated_bulk(context, results)
-
         for obj in objects:
             attrs = obj['attributes']
             if attrs and attrs.get(portbindings.HOST_ID):
@@ -1461,7 +1451,6 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
             LOG.error(_LE("mechanism_manager.delete_port_postcommit failed for"
                           " port %s"), port['id'])
         self.notifier.port_delete(context, port['id'])
-        self.notify_security_groups_member_updated(context, port)
 
     @utils.transaction_guard
     @db_api.retry_if_session_inactive(context_var_name='plugin_context')

neutron/tests/unit/plugins/ml2/test_plugin.py

@@ -889,7 +889,8 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
                                          'test', True, context=ctx)
             ports = self.deserialize(self.fmt, res)
             used_sg = ports['ports'][0]['security_groups']
-            m_upd.assert_called_once_with(ctx, used_sg)
+            m_upd.assert_has_calls(
+                [mock.call(ctx, [sg]) for sg in used_sg], any_order=True)
             self.assertFalse(p_upd.called)
 
     def _check_security_groups_provider_updated_args(self, p_upd_mock, net_id):