Merge "Send DHCP notifications regardless of agent status"

neutron/api/rpc/agentnotifiers/dhcp_rpc_agent_api.py

@@ -43,12 +43,11 @@ class DhcpAgentNotifyAPI(proxy.RpcProxy):
         super(DhcpAgentNotifyAPI, self).__init__(
             topic=topic, default_version=self.BASE_RPC_API_VERSION)
 
-    def _get_dhcp_agents(self, context, network_id):
+    def _get_enabled_dhcp_agents(self, context, network_id):
+        """Return enabled dhcp agents associated with the given network."""
         plugin = manager.NeutronManager.get_plugin()
-        dhcp_agents = plugin.get_dhcp_agents_hosting_networks(
-            context, [network_id], active=True)
-        return [(dhcp_agent.host, dhcp_agent.topic) for
-                dhcp_agent in dhcp_agents]
+        agents = plugin.get_dhcp_agents_hosting_networks(context, [network_id])
+        return [x for x in agents if x.admin_state_up]
 
     def _notification_host(self, context, method, payload, host):
         """Notify the agent on host."""
@@ -76,11 +75,33 @@ class DhcpAgentNotifyAPI(proxy.RpcProxy):
                             context, 'network_create_end',
                             {'network': {'id': network_id}},
                             agent['host'])
-            for (host, topic) in self._get_dhcp_agents(context, network_id):
+            agents = self._get_enabled_dhcp_agents(context, network_id)
+            if not agents:
+                LOG.error(_("No DHCP agents are associated with network "
+                            "'%(net_id)s'. Unable to send notification "
+                            "for '%(method)s' with payload: %(payload)s"),
+                          {
+                              'net_id': network_id,
+                              'method': method,
+                              'payload': payload,
+                          })
+                return
+            active_agents = [x for x in agents if x.is_active]
+            if active_agents != agents:
+                LOG.warning(_("Only %(active)d of %(total)d DHCP agents "
+                              "associated with network '%(net_id)s' are "
+                              "marked as active, so notifications may "
+                              "be sent to inactive agents."),
+                            {
+                                'active': len(active_agents),
+                                'total': len(agents),
+                                'net_id': network_id,
+                            })
+            for agent in agents:
                 self.cast(
                     context, self.make_msg(method,
                                            payload=payload),
-                    topic='%s.%s' % (topic, host))
+                    topic='%s.%s' % (agent.topic, agent.host))
         else:
             # besides the non-agentscheduler plugin,
             # There is no way to query who is hosting the network

neutron/db/agents_db.py

@@ -66,6 +66,10 @@ class Agent(model_base.BASEV2, models_v2.HasId):
     # configurations: a json dict string, I think 4095 is enough
     configurations = sa.Column(sa.String(4095), nullable=False)
 
+    @property
+    def is_active(self):
+        return not AgentDbMixin.is_agent_down(self.heartbeat_timestamp)
+
 
 class AgentDbMixin(ext_agent.AgentPluginBase):
     """Mixin class to add agent extension to db_plugin_base_v2."""

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

@@ -0,0 +1,76 @@
+# Copyright (c) 2013 Red Hat, Inc.
+#
+# 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 contextlib
+
+import mock
+
+from neutron.api.rpc.agentnotifiers import dhcp_rpc_agent_api
+from neutron.common import utils
+from neutron import manager
+from neutron.tests import base
+
+
+class TestDhcpAgentNotifyAPI(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestDhcpAgentNotifyAPI, self).setUp()
+        self.notify = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
+
+    def test_get_enabled_dhcp_agents_filters_disabled_agents(self):
+        disabled_agent = mock.Mock()
+        disabled_agent.admin_state_up = False
+        enabled_agent = mock.Mock()
+        with mock.patch.object(manager.NeutronManager,
+                               'get_plugin') as mock_get_plugin:
+            mock_get_plugin.return_value = mock_plugin = mock.Mock()
+            with mock.patch.object(
+                mock_plugin, 'get_dhcp_agents_hosting_networks'
+            ) as mock_get_agents:
+                mock_get_agents.return_value = [disabled_agent, enabled_agent]
+                result = self.notify._get_enabled_dhcp_agents('ctx', 'net_id')
+        self.assertEqual(result, [enabled_agent])
+
+    def _test_notification(self, agents):
+        with contextlib.nested(
+            mock.patch.object(manager.NeutronManager, 'get_plugin'),
+            mock.patch.object(utils, 'is_extension_supported'),
+            mock.patch.object(self.notify, '_get_enabled_dhcp_agents')
+        ) as (m1, m2, mock_get_agents):
+            mock_get_agents.return_value = agents
+            self.notify._notification(mock.Mock(), 'foo', {}, 'net_id')
+
+    def test_notification_sends_cast_for_enabled_agent(self):
+        with mock.patch.object(self.notify, 'cast') as mock_cast:
+            self._test_notification([mock.Mock()])
+        self.assertEqual(mock_cast.call_count, 1)
+
+    def test_notification_logs_error_for_no_enabled_agents(self):
+        with mock.patch.object(self.notify, 'cast') as mock_cast:
+            with mock.patch.object(dhcp_rpc_agent_api.LOG,
+                                   'error') as mock_log:
+                self._test_notification([])
+        self.assertEqual(mock_cast.call_count, 0)
+        self.assertEqual(mock_log.call_count, 1)
+
+    def test_notification_logs_warning_for_inactive_agents(self):
+        agent = mock.Mock()
+        agent.is_active = False
+        with mock.patch.object(self.notify, 'cast') as mock_cast:
+            with mock.patch.object(dhcp_rpc_agent_api.LOG,
+                                   'warning') as mock_log:
+                self._test_notification([agent])
+        self.assertEqual(mock_cast.call_count, 1)
+        self.assertEqual(mock_log.call_count, 1)