Add LeastRouters Scheduler to Neutron L3 Agent

Allow scheduling of a virtual router on an L3 Agent node with the least
number of routers currently scheduled. This scheduler can be used
instead of the default random scheduler.

Also refactor the l3_agent_scheduler to allow for adding new schedulers.

Implement blueprint lessrouter-scheduler
Change-Id: Ie539c08bdc8a6e1430a106f77d08f15abd0903e7

neutron/db/l3_agentschedulers_db.py

@@ -17,6 +17,7 @@
 
 from oslo.config import cfg
 import sqlalchemy as sa
+from sqlalchemy import func
 from sqlalchemy import orm
 from sqlalchemy.orm import exc
 from sqlalchemy.orm import joinedload
@@ -249,3 +250,14 @@ class L3AgentSchedulerDbMixin(l3agentscheduler.L3AgentSchedulerPluginBase,
         """Schedule the routers to l3 agents."""
         for router in routers:
             self.schedule_router(context, router)
+
+    def get_l3_agent_with_min_routers(self, context, agent_ids):
+        """Return l3 agent with the least number of routers."""
+        query = context.session.query(
+            agents_db.Agent,
+            func.count(
+                RouterL3AgentBinding.router_id
+            ).label('count')).outerjoin(RouterL3AgentBinding).group_by(
+                RouterL3AgentBinding.l3_agent_id).order_by('count')
+        res = query.filter(agents_db.Agent.id.in_(agent_ids)).first()
+        return res[0]

neutron/scheduler/l3_agent_scheduler.py

@@ -15,8 +15,10 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import abc
 import random
 
+import six
 from sqlalchemy.orm import exc
 from sqlalchemy.sql import exists
 
@@ -30,14 +32,20 @@ from neutron.openstack.common import log as logging
 LOG = logging.getLogger(__name__)
 
 
-class ChanceScheduler(object):
+@six.add_metaclass(abc.ABCMeta)
-    """Allocate a L3 agent for a router in a random way.
+class L3Scheduler(object):
-    More sophisticated scheduler (similar to filter scheduler in nova?)
+
-    can be introduced later.
+    @abc.abstractmethod
+    def schedule(self, plugin, context, router_id):
+        """Schedule the router to an active L3 agent.
+
+        Schedule the router only if it is not already scheduled.
         """
+        pass
 
     def auto_schedule_routers(self, plugin, context, host, router_ids):
         """Schedule non-hosted routers to L3 Agent running on host.
+
         If router_ids is given, each router in router_ids is scheduled
         if it is not scheduled yet. Otherwise all unscheduled routers
         are scheduled.
@@ -104,34 +112,26 @@ class ChanceScheduler(object):
                            ' on host %s'), host)
                 return False
 
-            # binding
             for router_id in router_ids:
-                binding = l3_agentschedulers_db.RouterL3AgentBinding()
+                self.bind_router(context, router_id, l3_agent)
-                binding.l3_agent = l3_agent
-                binding.router_id = router_id
-                binding.default = True
-                context.session.add(binding)
         return True
 
-    def schedule(self, plugin, context, router_id):
+    def get_candidates(self, plugin, context, sync_router):
-        """Schedule the router to an active L3 agent if there
+        """Return L3 agents where a router could be scheduled."""
-        is no enable L3 agent hosting it.
-        """
         with context.session.begin(subtransactions=True):
             # allow one router is hosted by just
             # one enabled l3 agent hosting since active is just a
             # timing problem. Non-active l3 agent can return to
             # active any time
             l3_agents = plugin.get_l3_agents_hosting_routers(
-                context, [router_id], admin_state_up=True)
+                context, [sync_router['id']], admin_state_up=True)
             if l3_agents:
                 LOG.debug(_('Router %(router_id)s has already been hosted'
                             ' by L3 agent %(agent_id)s'),
-                          {'router_id': router_id,
+                          {'router_id': sync_router['id'],
                            'agent_id': l3_agents[0]['id']})
                 return
 
-            sync_router = plugin.get_router(context, router_id)
             active_l3_agents = plugin.get_l3_agents(context, active=True)
             if not active_l3_agents:
                 LOG.warn(_('No active L3 agents'))
@@ -143,13 +143,50 @@ class ChanceScheduler(object):
                          sync_router['id'])
                 return
 
-            chosen_agent = random.choice(candidates)
+            return candidates
+
+    def bind_router(self, context, router_id, chosen_agent):
+        """Bind the router to the l3 agent which has been chosen."""
+        with context.session.begin(subtransactions=True):
             binding = l3_agentschedulers_db.RouterL3AgentBinding()
             binding.l3_agent = chosen_agent
-            binding.router_id = sync_router['id']
+            binding.router_id = router_id
             context.session.add(binding)
             LOG.debug(_('Router %(router_id)s is scheduled to '
                         'L3 agent %(agent_id)s'),
-                      {'router_id': sync_router['id'],
+                      {'router_id': router_id,
-                       'agent_id': chosen_agent['id']})
+                       'agent_id': chosen_agent.id})
+
+
+class ChanceScheduler(L3Scheduler):
+    """Randomly allocate an L3 agent for a router."""
+
+    def schedule(self, plugin, context, router_id):
+        with context.session.begin(subtransactions=True):
+            sync_router = plugin.get_router(context, router_id)
+            candidates = self.get_candidates(plugin, context, sync_router)
+            if not candidates:
+                return
+
+            chosen_agent = random.choice(candidates)
+            self.bind_router(context, router_id, chosen_agent)
+            return chosen_agent
+
+
+class LeastRoutersScheduler(L3Scheduler):
+    """Allocate to an L3 agent with the least number of routers bound."""
+
+    def schedule(self, plugin, context, router_id):
+        with context.session.begin(subtransactions=True):
+            sync_router = plugin.get_router(context, router_id)
+            candidates = self.get_candidates(plugin, context, sync_router)
+            if not candidates:
+                return
+
+            candidate_ids = [candidate['id'] for candidate in candidates]
+            chosen_agent = plugin.get_l3_agent_with_min_routers(
+                context, candidate_ids)
+
+            self.bind_router(context, router_id, chosen_agent)
+
             return chosen_agent

neutron/tests/unit/test_l3_schedulers.py

@@ -0,0 +1,210 @@
+# 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.
+#
+# @author: Sylvain Afchain, eNovance SAS
+# @author: Emilien Macchi, eNovance SAS
+
+import contextlib
+import uuid
+
+import mock
+from oslo.config import cfg
+
+from neutron.api.v2 import attributes as attr
+from neutron.common import constants
+from neutron.common.test_lib import test_config
+from neutron.common import topics
+from neutron import context as q_context
+from neutron.db import agents_db
+from neutron.db import l3_agentschedulers_db
+from neutron.extensions import l3 as ext_l3
+from neutron import manager
+from neutron.openstack.common import timeutils
+from neutron.tests.unit import test_db_plugin
+from neutron.tests.unit import test_l3_plugin
+
+HOST = 'my_l3_host'
+FIRST_L3_AGENT = {
+    'binary': 'neutron-l3-agent',
+    'host': HOST,
+    'topic': topics.L3_AGENT,
+    'configurations': {},
+    'agent_type': constants.AGENT_TYPE_L3,
+    'start_flag': True
+}
+
+HOST_2 = 'my_l3_host_2'
+SECOND_L3_AGENT = {
+    'binary': 'neutron-l3-agent',
+    'host': HOST_2,
+    'topic': topics.L3_AGENT,
+    'configurations': {},
+    'agent_type': constants.AGENT_TYPE_L3,
+    'start_flag': True
+}
+
+DB_PLUGIN_KLASS = ('neutron.plugins.openvswitch.ovs_neutron_plugin.'
+                   'OVSNeutronPluginV2')
+
+
+class L3SchedulerTestExtensionManager(object):
+
+    def get_resources(self):
+        attr.RESOURCE_ATTRIBUTE_MAP.update(ext_l3.RESOURCE_ATTRIBUTE_MAP)
+        l3_res = ext_l3.L3.get_resources()
+        return l3_res
+
+    def get_actions(self):
+        return []
+
+    def get_request_extensions(self):
+        return []
+
+
+class L3SchedulerTestCase(l3_agentschedulers_db.L3AgentSchedulerDbMixin,
+                          test_db_plugin.NeutronDbPluginV2TestCase,
+                          test_l3_plugin.L3NatTestCaseMixin):
+
+    def setUp(self):
+        test_config['plugin_name_v2'] = DB_PLUGIN_KLASS
+
+        ext_mgr = L3SchedulerTestExtensionManager()
+        test_config['extension_manager'] = ext_mgr
+
+        super(L3SchedulerTestCase, self).setUp()
+
+        self.adminContext = q_context.get_admin_context()
+        self.plugin = manager.NeutronManager.get_plugin()
+        self._register_l3_agents()
+
+    def _register_l3_agents(self):
+        callback = agents_db.AgentExtRpcCallback()
+        callback.report_state(self.adminContext,
+                              agent_state={'agent_state': FIRST_L3_AGENT},
+                              time=timeutils.strtime())
+        agent_db = self.plugin.get_agents_db(self.adminContext,
+                                             filters={'host': [HOST]})
+        self.agent_id1 = agent_db[0].id
+
+        callback.report_state(self.adminContext,
+                              agent_state={'agent_state': SECOND_L3_AGENT},
+                              time=timeutils.strtime())
+        agent_db = self.plugin.get_agents_db(self.adminContext,
+                                             filters={'host': [HOST]})
+        self.agent_id2 = agent_db[0].id
+
+    def _set_l3_agent_admin_state(self, context, agent_id, state=True):
+        update = {'agent': {'admin_state_up': state}}
+        self.plugin.update_agent(context, agent_id, update)
+
+    @contextlib.contextmanager
+    def router_with_ext_gw(self, name='router1', admin_state_up=True,
+                           fmt=None, tenant_id=str(uuid.uuid4()),
+                           external_gateway_info=None,
+                           subnet=None, set_context=False,
+                           **kwargs):
+        router = self._make_router(fmt or self.fmt, tenant_id, name,
+                                   admin_state_up, external_gateway_info,
+                                   set_context, **kwargs)
+        self._add_external_gateway_to_router(
+            router['router']['id'],
+            subnet['subnet']['network_id'])
+        try:
+            yield router
+        finally:
+            self._remove_external_gateway_from_router(
+                router['router']['id'], subnet['subnet']['network_id'])
+            self._delete('routers', router['router']['id'])
+
+
+class L3AgentChanceSchedulerTestCase(L3SchedulerTestCase):
+
+    def test_random_scheduling(self):
+        random_patch = mock.patch('random.choice')
+        random_mock = random_patch.start()
+
+        def side_effect(seq):
+            return seq[0]
+        random_mock.side_effect = side_effect
+
+        with self.subnet() as subnet:
+            self._set_net_external(subnet['subnet']['network_id'])
+            with self.router_with_ext_gw(name='r1', subnet=subnet) as r1:
+                agents = self.get_l3_agents_hosting_routers(
+                    self.adminContext, [r1['router']['id']],
+                    admin_state_up=True)
+
+                self.assertEqual(len(agents), 1)
+                self.assertEqual(random_mock.call_count, 1)
+
+                with self.router_with_ext_gw(name='r2', subnet=subnet) as r2:
+                    agents = self.get_l3_agents_hosting_routers(
+                        self.adminContext, [r2['router']['id']],
+                        admin_state_up=True)
+
+                    self.assertEqual(len(agents), 1)
+                    self.assertEqual(random_mock.call_count, 2)
+
+        random_patch.stop()
+
+
+class L3AgentLeastRoutersSchedulerTestCase(L3SchedulerTestCase):
+    def setUp(self):
+        cfg.CONF.set_override('router_scheduler_driver',
+                              'neutron.scheduler.l3_agent_scheduler.'
+                              'LeastRoutersScheduler')
+
+        super(L3AgentLeastRoutersSchedulerTestCase, self).setUp()
+
+    def test_scheduler(self):
+        # disable one agent to force the scheduling to the only one.
+        self._set_l3_agent_admin_state(self.adminContext,
+                                       self.agent_id2, False)
+
+        with self.subnet() as subnet:
+            self._set_net_external(subnet['subnet']['network_id'])
+            with self.router_with_ext_gw(name='r1', subnet=subnet) as r1:
+                agents = self.get_l3_agents_hosting_routers(
+                    self.adminContext, [r1['router']['id']],
+                    admin_state_up=True)
+                self.assertEqual(len(agents), 1)
+
+                agent_id1 = agents[0]['id']
+
+                with self.router_with_ext_gw(name='r2', subnet=subnet) as r2:
+                    agents = self.get_l3_agents_hosting_routers(
+                        self.adminContext, [r2['router']['id']],
+                        admin_state_up=True)
+                    self.assertEqual(len(agents), 1)
+
+                    agent_id2 = agents[0]['id']
+
+                    self.assertEqual(agent_id1, agent_id2)
+
+                    # re-enable the second agent to see whether the next router
+                    # spawned will be on this one.
+                    self._set_l3_agent_admin_state(self.adminContext,
+                                                   self.agent_id2, True)
+
+                    with self.router_with_ext_gw(name='r3',
+                                                 subnet=subnet) as r3:
+                        agents = self.get_l3_agents_hosting_routers(
+                            self.adminContext, [r3['router']['id']],
+                            admin_state_up=True)
+                        self.assertEqual(len(agents), 1)
+
+                        agent_id3 = agents[0]['id']
+
+                        self.assertNotEqual(agent_id1, agent_id3)