L3 DVR: use fanout when sending dvr arp table update

Sending arp update to each l3 dvr agent one by one on every port
creation is not scalable and causes serious performance degradation
if router is hosted on lots of l3 dvr agents on compute nodes (see
bug report). This increases port creation time and eventually leads
to timeouts in Nova and VMs going to ERROR state.

This patch changes notification to be fanout.
The downside is that with fanout the arp notification will be sent to
each l3 agent, even those not hosting the router. However such agents
will just skip the notification if not hosting the router - this should
be quite cheap.

Closes-Bug: #1614452
Change-Id: I1fb533d7804b131f709b790fc730ed7b97cb5499

neutron/api/rpc/agentnotifiers/l3_rpc_agent_api.py

@@ -79,23 +79,10 @@ class L3AgentNotifyAPI(object):
         """Notify arp details to l3 agents hosting router."""
         if not router_id:
             return
-        adminContext = (context.is_admin and
-                        context or context.elevated())
-        plugin = manager.NeutronManager.get_service_plugins().get(
-            service_constants.L3_ROUTER_NAT)
-        hosts = plugin.get_hosts_to_notify(adminContext, router_id)
-        # TODO(murali): replace cast with fanout to avoid performance
-        # issues at greater scale.
-        for host in hosts:
-            log_topic = '%s.%s' % (topics.L3_AGENT, host)
-            LOG.debug('Casting message %(method)s with topic %(topic)s',
-                      {'topic': log_topic, 'method': method})
-            dvr_arptable = {'router_id': router_id,
-                            'arp_table': data}
-            cctxt = self.client.prepare(topic=topics.L3_AGENT,
-                                        server=host,
-                                        version='1.2')
-            cctxt.cast(context, method, payload=dvr_arptable)
+        dvr_arptable = {'router_id': router_id, 'arp_table': data}
+        LOG.debug('Fanout dvr_arptable update: %s', dvr_arptable)
+        cctxt = self.client.prepare(fanout=True, version='1.2')
+        cctxt.cast(context, method, payload=dvr_arptable)
 
     def _notification(self, context, method, router_ids, operation,
                       shuffle_agents, schedule_routers=True):

neutron/tests/unit/api/rpc/agentnotifiers/test_l3_rpc_agent_api.py

@@ -0,0 +1,47 @@
+# Copyright (c) 2016 OpenStack Foundation.
+#
+# 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 mock
+
+from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
+from neutron.tests import base
+
+
+class TestL3AgentNotifyAPI(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestL3AgentNotifyAPI, self).setUp()
+        self.rpc_client_mock = mock.patch(
+            'neutron.common.rpc.get_client').start().return_value
+        self.l3_notifier = l3_rpc_agent_api.L3AgentNotifyAPI()
+
+    def _test_arp_update(self, method):
+        arp_table = {'ip_address': '1.1.1.1',
+                     'mac_address': '22:f1:6c:9c:79:4a',
+                     'subnet_id': 'subnet_id'}
+        router_id = 'router_id'
+        getattr(self.l3_notifier, method)(mock.Mock(), router_id, arp_table)
+        self.rpc_client_mock.prepare.assert_called_once_with(
+            fanout=True, version='1.2')
+        cctxt = self.rpc_client_mock.prepare.return_value
+        cctxt.cast.assert_called_once_with(
+            mock.ANY, method,
+            payload={'router_id': router_id, 'arp_table': arp_table})
+
+    def test_add_arp_entry(self):
+        self._test_arp_update('add_arp_entry')
+
+    def test_del_arp_entry(self):
+        self._test_arp_update('del_arp_entry')