Finish Agent Scheduler and LB Agent Driver Piece

Implements the agent scheduler to choose which agent will host
the loadbalancer.  Only a random scheduling algorithm implemented.

Exposed two API calls that allows an admin user to get a list of
loadbalancers hosted by an agent and get the agent that is hosting
a particular loadbalancer.

Implemented the loadbalancer driver classes and methods for the agent
driver.  These will call the agent scheduler to host the load balancer
and then send the data to a queue.

Depends-On: Ic9179ef0a95b91d1b7662537fffeb0a949efc925
Partially-implements: blueprint lbaas-refactor-haproxy-namespace-driver-to-new-driver-interface
Co-Authored-By: Brandon Logan <brandon.logan@rackspace.com>

Change-Id: Ic4a83ef255aed0474eebb73f28d67e09f35666b7

etc/neutron_lbaas.conf

@@ -1,3 +1,8 @@
+[DEFAULT]
+# =========== items for agent scheduler extension =============
+# loadbalancer_pool_scheduler_driver = neutron.services.loadbalancer.agent_scheduler.ChanceScheduler
+# loadbalancer_scheduler_driver = neutron.agent_scheduler.ChanceScheduler
+
 [quotas]
 # Number of vips allowed per tenant. A negative value means unlimited.  This
 # is only applicable when v1 of the lbaas extension is used.

neutron_lbaas/agent_scheduler.py

@@ -0,0 +1,143 @@
+# 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 random
+
+from neutron.db import agents_db
+from neutron.db import agentschedulers_db
+from neutron.db import model_base
+from neutron.i18n import _LW
+from neutron.openstack.common import log as logging
+import sqlalchemy as sa
+from sqlalchemy import orm
+from sqlalchemy.orm import joinedload
+
+from neutron_lbaas.extensions import lbaas_agentschedulerv2
+from neutron_lbaas.services.loadbalancer import constants as lb_const
+
+LOG = logging.getLogger(__name__)
+
+
+class LoadbalancerAgentBinding(model_base.BASEV2):
+    """Represents binding between neutron loadbalancer and agents."""
+
+    __tablename__ = "lbaas_loadbalanceragentbindings"
+
+    loadbalancer_id = sa.Column(
+        sa.String(36),
+        sa.ForeignKey("lbaas_loadbalancers.id", ondelete='CASCADE'),
+        primary_key=True)
+    agent = orm.relation(agents_db.Agent)
+    agent_id = sa.Column(
+        sa.String(36),
+        sa.ForeignKey("agents.id", ondelete='CASCADE'),
+        nullable=False)
+
+
+class LbaasAgentSchedulerDbMixin(agentschedulers_db.AgentSchedulerDbMixin,
+                                 lbaas_agentschedulerv2
+                                 .LbaasAgentSchedulerPluginBase):
+
+    agent_notifiers = {}
+
+    def get_agent_hosting_loadbalancer(self, context,
+                                       loadbalancer_id, active=None):
+        query = context.session.query(LoadbalancerAgentBinding)
+        query = query.options(joinedload('agent'))
+        binding = query.get(loadbalancer_id)
+
+        if (binding and self.is_eligible_agent(
+                active, binding.agent)):
+            return {'agent': self._make_agent_dict(binding.agent)}
+
+    def get_lbaas_agents(self, context, active=None, filters=None):
+        query = context.session.query(agents_db.Agent)
+        query = query.filter_by(agent_type=lb_const.AGENT_TYPE_LOADBALANCERV2)
+        if active is not None:
+            query = query.filter_by(admin_state_up=active)
+        if filters:
+            for key, value in filters.iteritems():
+                column = getattr(agents_db.Agent, key, None)
+                if column:
+                    query = query.filter(column.in_(value))
+
+        return [agent
+                for agent in query
+                if self.is_eligible_agent(active, agent)]
+
+    def list_loadbalancers_on_lbaas_agent(self, context, id):
+        query = context.session.query(
+            LoadbalancerAgentBinding.loadbalancer_id)
+        query = query.filter_by(agent_id=id)
+        loadbalancer_ids = [item[0] for item in query]
+        if loadbalancer_ids:
+            lbs = self.get_loadbalancers(context,
+                                         filters={'id': loadbalancer_ids})
+            return lbs
+        return []
+
+    def get_lbaas_agent_candidates(self, device_driver, active_agents):
+        candidates = []
+        for agent in active_agents:
+            agent_conf = self.get_configuration_dict(agent)
+            if device_driver in agent_conf['device_drivers']:
+                candidates.append(agent)
+        return candidates
+
+
+class ChanceScheduler(object):
+    """Allocate a loadbalancer agent for a vip in a random way."""
+
+    def schedule(self, plugin, context, loadbalancer, device_driver):
+        """Schedule the load balancer to an active loadbalancer agent if there
+        is no enabled agent hosting it.
+        """
+        with context.session.begin(subtransactions=True):
+            lbaas_agent = plugin.db.get_agent_hosting_loadbalancer(
+                context, loadbalancer.id)
+            if lbaas_agent:
+                LOG.debug('Load balancer %(loadbalancer_id)s '
+                          'has already been hosted'
+                          ' by lbaas agent %(agent_id)s',
+                          {'loadbalancer_id': loadbalancer.id,
+                           'agent_id': lbaas_agent['id']})
+                return
+
+            active_agents = plugin.db.get_lbaas_agents(context, active=True)
+            if not active_agents:
+                LOG.warn(
+                    _LW('No active lbaas agents for load balancer %s'),
+                    loadbalancer.id)
+                return
+
+            candidates = plugin.db.get_lbaas_agent_candidates(device_driver,
+                                                              active_agents)
+            if not candidates:
+                LOG.warn(_LW('No lbaas agent supporting device driver %s'),
+                         device_driver)
+                return
+
+            chosen_agent = random.choice(candidates)
+            binding = LoadbalancerAgentBinding()
+            binding.agent = chosen_agent
+            binding.loadbalancer_id = loadbalancer.id
+            context.session.add(binding)
+            LOG.debug(
+                'Load balancer %(loadbalancer_id)s is scheduled '
+                'to lbaas agent %(agent_id)s', {
+                    'loadbalancer_id': loadbalancer.id,
+                    'agent_id': chosen_agent['id']}
+            )
+            return chosen_agent

neutron_lbaas/db/loadbalancer/loadbalancer_dbv2.py

@@ -24,6 +24,7 @@ from oslo_utils import uuidutils
 from sqlalchemy import orm
 from sqlalchemy.orm import exc
 
+from neutron_lbaas import agent_scheduler
 from neutron_lbaas.db.loadbalancer import models
 from neutron_lbaas.extensions import loadbalancerv2
 from neutron_lbaas.services.loadbalancer import constants as lb_const
@@ -33,7 +34,8 @@ from neutron_lbaas.services.loadbalancer import data_models
 LOG = logging.getLogger(__name__)
 
 
-class LoadBalancerPluginDbv2(base_db.CommonDbMixin):
+class LoadBalancerPluginDbv2(base_db.CommonDbMixin,
+                             agent_scheduler.LbaasAgentSchedulerDbMixin):
     """Wraps loadbalancer with SQLAlchemy models.
 
     A class that wraps the implementation of the Neutron loadbalancer

neutron_lbaas/drivers/common/agent_driver_base.py

@@ -0,0 +1,176 @@
+# Copyright 2015 Rackspace.
+#
+#    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 neutron.common import exceptions as n_exc
+from neutron.common import rpc as n_rpc
+from neutron.db import agents_db
+from neutron.openstack.common import log as logging
+from neutron.services import provider_configuration as provconf
+from oslo_config import cfg
+import oslo_messaging as messaging
+from oslo_utils import importutils
+
+from neutron_lbaas.drivers import driver_base
+from neutron_lbaas.extensions import lbaas_agentschedulerv2
+from neutron_lbaas.services.loadbalancer import constants as lb_const
+from neutron_lbaas.services.loadbalancer import data_models
+
+LOG = logging.getLogger(__name__)
+
+LB_SCHEDULERS = 'loadbalancer_schedulers'
+
+AGENT_SCHEDULER_OPTS = [
+    cfg.StrOpt('loadbalancer_scheduler_driver',
+               default='neutron_lbaas.agent_scheduler.ChanceScheduler',
+               help=_('Driver to use for scheduling '
+                      'to a default loadbalancer agent')),
+]
+
+cfg.CONF.register_opts(AGENT_SCHEDULER_OPTS)
+
+
+class DriverNotSpecified(n_exc.NeutronException):
+    message = _("Device driver for agent should be specified "
+                "in plugin driver.")
+
+
+class DataModelSerializer(object):
+
+    def serialize_entity(self, ctx, entity):
+        if isinstance(entity, data_models.BaseDataModel):
+            return entity.to_dict(stats=False)
+        else:
+            return entity
+
+
+class LoadBalancerAgentApi(object):
+    """Plugin side of plugin to agent RPC API."""
+
+    # history
+    #   1.0 Initial version
+    #
+
+    def __init__(self, topic):
+        target = messaging.Target(topic=topic, version='1.0')
+        self.client = n_rpc.get_client(target,
+                                       serializer=DataModelSerializer())
+
+    def agent_updated(self, context, admin_state_up, host):
+        cctxt = self.client.prepare(server=host)
+        cctxt.cast(context, 'agent_updated',
+                   payload={'admin_state_up': admin_state_up})
+
+    def create_loadbalancer(self, context, loadbalancer, host, driver_name):
+        cctxt = self.client.prepare(server=host)
+        cctxt.cast(context, 'create_loadbalancer',
+                   loadbalancer=loadbalancer, driver_name=driver_name)
+
+    def update_loadbalancer(self, context, old_loadbalancer,
+                            loadbalancer, host):
+        cctxt = self.client.prepare(server=host)
+        cctxt.cast(context, 'update_loadbalancer',
+                   old_loadbalancer=old_loadbalancer,
+                   loadbalancer=loadbalancer)
+
+    def delete_loadbalancer(self, context, loadbalancer, host):
+        cctxt = self.client.prepare(server=host)
+        cctxt.cast(context, 'delete_loadbalancer', loadbalancer=loadbalancer)
+
+
+class LoadBalancerManager(driver_base.BaseLoadBalancerManager):
+
+    def update(self, context, old_loadbalancer, loadbalancer):
+        super(LoadBalancerManager, self).update(context, old_loadbalancer,
+                                                loadbalancer)
+        agent = self.driver.get_loadbalancer_agent(context, loadbalancer.id)
+        self.driver.agent_rpc.update_loadbalancer(
+            context, old_loadbalancer, loadbalancer, agent['host'])
+
+    def create(self, context, loadbalancer):
+        super(LoadBalancerManager, self).create(context, loadbalancer)
+        agent = self.driver.loadbalancer_scheduler.schedule(
+            self.driver.plugin, context, loadbalancer,
+            self.driver.device_driver)
+        if not agent:
+            raise lbaas_agentschedulerv2.NoEligibleLbaasAgent(
+                loadbalancer_id=loadbalancer.id)
+        self.driver.agent_rpc.create_loadbalancer(
+            context, loadbalancer, agent['host'], self.driver.device_driver)
+
+    def delete(self, context, loadbalancer):
+        super(LoadBalancerManager, self).delete(context, loadbalancer)
+        agent = self.driver.get_loadbalancer_agent(context, loadbalancer.id)
+        # TODO(blogan): Rethink deleting from the database here. May want to
+        # wait until the agent actually deletes it.  Doing this now to keep
+        # what v1 had.
+        self.driver.plugin.db.delete_loadbalancer(context, loadbalancer.id)
+        if agent:
+            self.driver.agent_rpc.delete_loadbalancer(context, loadbalancer,
+                                                      agent['host'])
+
+    def stats(self, context, loadbalancer):
+        pass
+
+    def refresh(self, context, loadbalancer):
+        pass
+
+
+class AgentDriverBase(driver_base.LoadBalancerBaseDriver):
+
+    # name of device driver that should be used by the agent;
+    # vendor specific plugin drivers must override it;
+    device_driver = None
+
+    def __init__(self, plugin):
+        super(AgentDriverBase, self).__init__(plugin)
+        if not self.device_driver:
+            raise DriverNotSpecified()
+
+        self.load_balancer = LoadBalancerManager(self)
+
+        self.agent_rpc = LoadBalancerAgentApi(lb_const.LOADBALANCER_AGENTV2)
+
+        self._set_callbacks_on_plugin()
+        # Setting this on the db because the plugin no longer inherts from
+        # database classes, the db does.
+        self.plugin.db.agent_notifiers.update(
+            {lb_const.AGENT_TYPE_LOADBALANCERV2: self.agent_rpc})
+
+        lb_sched_driver = provconf.get_provider_driver_class(
+            cfg.CONF.loadbalancer_scheduler_driver, LB_SCHEDULERS)
+        self.loadbalancer_scheduler = importutils.import_object(
+            lb_sched_driver)
+
+    def _set_callbacks_on_plugin(self):
+        # other agent based plugin driver might already set callbacks on plugin
+        if hasattr(self.plugin, 'agent_callbacks'):
+            return
+
+        self.plugin.agent_endpoints = [
+            agents_db.AgentExtRpcCallback(self.plugin.db)
+        ]
+        self.plugin.conn = n_rpc.create_connection(new=True)
+        self.plugin.conn.create_consumer(
+            lb_const.LOADBALANCER_PLUGINV2,
+            self.plugin.agent_endpoints,
+            fanout=False)
+        self.plugin.conn.consume_in_threads()
+
+    def get_loadbalancer_agent(self, context, loadbalancer_id):
+        agent = self.plugin.db.get_agent_hosting_loadbalancer(
+            context, loadbalancer_id)
+        if not agent:
+            raise lbaas_agentschedulerv2.NoActiveLbaasAgent(
+                loadbalancer_id=loadbalancer_id)
+        return agent['agent']

neutron_lbaas/drivers/haproxy/plugin_driver.py

@@ -0,0 +1,20 @@
+# Copyright (c) 2015 Rackspace.
+# 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 neutron_lbaas.drivers.common import agent_driver_base
+
+
+class HaproxyOnHostPluginDriver(agent_driver_base.AgentDriverBase):
+    device_driver = 'test'

neutron_lbaas/services/loadbalancer/plugin.py

@@ -25,6 +25,7 @@ from neutron.services import service_base
 from oslo_config import cfg
 from oslo_utils import excutils
 
+from neutron_lbaas import agent_scheduler as agent_scheduler_v2
 from neutron_lbaas.db.loadbalancer import loadbalancer_db as ldb
 from neutron_lbaas.db.loadbalancer import loadbalancer_dbv2 as ldbv2
 from neutron_lbaas.db.loadbalancer import models
@@ -368,7 +369,8 @@ class LoadBalancerPluginv2(loadbalancerv2.LoadBalancerPluginBaseV2):
                                    "lbaas_agent_schedulerv2",
                                    "service-type"]
 
-    agent_notifiers = {}
+    agent_notifiers = (
+        agent_scheduler_v2.LbaasAgentSchedulerDbMixin.agent_notifiers)
 
     def __init__(self):
         """Initialization for the loadbalancer service plugin."""

neutron_lbaas/tests/unit/db/loadbalancer/test_db_loadbalancerv2.py

@@ -15,6 +15,7 @@
 
 import contextlib
 
+import mock
 from neutron.api import extensions
 from neutron.api.v2 import attributes
 from neutron.common import config
@@ -324,6 +325,13 @@ class LbaasPluginDbTestCase(LbaasTestMixin, base.NeutronDbPluginV2TestCase):
             app = config.load_paste_app('extensions_test_app')
             self.ext_api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
 
+        get_lbaas_agent_patcher = mock.patch(
+            'neutron_lbaas.agent_scheduler'
+            '.LbaasAgentSchedulerDbMixin.get_agent_hosting_loadbalancer')
+        mock_lbaas_agent = mock.MagicMock()
+        get_lbaas_agent_patcher.start().return_value = mock_lbaas_agent
+        mock_lbaas_agent.__getitem__.return_value = {'host': 'host'}
+
         self._subnet_id = _subnet_id
 
     def _update_loadbalancer_api(self, lb_id, data):

neutron_lbaas/tests/unit/drivers/common/test_agent_driver_base.py

@@ -0,0 +1,168 @@
+# Copyright 2013 New Dream Network, LLC (DreamHost)
+#
+#    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 import context
+from neutron.db import servicetype_db as st_db
+from neutron import manager
+from neutron.plugins.common import constants
+
+from neutron_lbaas.db.loadbalancer import models
+from neutron_lbaas.drivers.common import agent_driver_base
+from neutron_lbaas.extensions import loadbalancerv2
+from neutron_lbaas.tests import base
+from neutron_lbaas.tests.unit.db.loadbalancer import test_db_loadbalancerv2
+
+
+class TestLoadBalancerPluginBase(test_db_loadbalancerv2.LbaasPluginDbTestCase):
+
+    def setUp(self):
+        def reset_device_driver():
+            agent_driver_base.AgentDriverBase.device_driver = None
+        self.addCleanup(reset_device_driver)
+
+        self.mock_importer = mock.patch.object(
+            agent_driver_base, 'importutils').start()
+
+        # needed to reload provider configuration
+        st_db.ServiceTypeManager._instance = None
+        agent_driver_base.AgentDriverBase.device_driver = 'dummy'
+        super(TestLoadBalancerPluginBase, self).setUp(
+            lbaas_provider=('LOADBALANCERV2:lbaas:neutron_lbaas.drivers.'
+                            'common.agent_driver_base.'
+                            'AgentDriverBase:default'))
+
+        # we need access to loaded plugins to modify models
+        loaded_plugins = manager.NeutronManager().get_service_plugins()
+
+        self.plugin_instance = loaded_plugins[constants.LOADBALANCERV2]
+
+
+class TestLoadBalancerAgentApi(base.BaseTestCase):
+    def setUp(self):
+        super(TestLoadBalancerAgentApi, self).setUp()
+
+        self.api = agent_driver_base.LoadBalancerAgentApi('topic')
+
+    def test_init(self):
+        self.assertEqual(self.api.client.target.topic, 'topic')
+
+    def _call_test_helper(self, method_name, method_args):
+        with contextlib.nested(
+            mock.patch.object(self.api.client, 'cast'),
+            mock.patch.object(self.api.client, 'prepare'),
+        ) as (
+            rpc_mock, prepare_mock
+        ):
+            prepare_mock.return_value = self.api.client
+            getattr(self.api, method_name)(mock.sentinel.context,
+                                           host='host',
+                                           **method_args)
+
+        prepare_args = {'server': 'host'}
+        prepare_mock.assert_called_once_with(**prepare_args)
+
+        if method_name == 'agent_updated':
+            method_args = {'payload': method_args}
+        rpc_mock.assert_called_once_with(mock.sentinel.context, method_name,
+                                         **method_args)
+
+    def test_agent_updated(self):
+        self._call_test_helper('agent_updated', {'admin_state_up': 'test'})
+
+    def test_create_loadbalancer(self):
+        self._call_test_helper('create_loadbalancer', {'loadbalancer': 'test',
+                                                       'driver_name': 'dummy'})
+
+    def test_update_loadbalancer(self):
+        self._call_test_helper('update_loadbalancer', {
+            'old_loadbalancer': 'test', 'loadbalancer': 'test'})
+
+    def test_delete_loadbalancer(self):
+        self._call_test_helper('delete_loadbalancer', {'loadbalancer': 'test'})
+
+
+class TestLoadBalancerPluginNotificationWrapper(TestLoadBalancerPluginBase):
+    def setUp(self):
+        self.log = mock.patch.object(agent_driver_base, 'LOG')
+        api_cls = mock.patch.object(agent_driver_base,
+                                    'LoadBalancerAgentApi').start()
+        super(TestLoadBalancerPluginNotificationWrapper, self).setUp()
+        self.mock_api = api_cls.return_value
+
+        self.mock_get_driver = mock.patch.object(self.plugin_instance,
+                                                 '_get_driver')
+        self.mock_get_driver.return_value = (
+            agent_driver_base.AgentDriverBase(self.plugin_instance))
+
+    def _update_status(self, model, status, id):
+        ctx = context.get_admin_context()
+        self.plugin_instance.db.update_status(
+            ctx,
+            model,
+            id,
+            provisioning_status=status
+        )
+
+    def test_create_loadbalancer(self):
+        with self.loadbalancer(no_delete=True) as loadbalancer:
+            calls = self.mock_api.create_loadbalancer.call_args_list
+            self.assertEqual(1, len(calls))
+            _, called_lb, _, device_driver = calls[0][0]
+            self.assertEqual(loadbalancer['loadbalancer']['id'], called_lb.id)
+            self.assertEqual('dummy', device_driver)
+            self.assertEqual(constants.PENDING_CREATE,
+                             called_lb.provisioning_status)
+
+    def test_update_loadbalancer(self):
+        with self.loadbalancer(no_delete=True) as loadbalancer:
+            lb_id = loadbalancer['loadbalancer']['id']
+            old_lb_name = loadbalancer['loadbalancer']['name']
+            ctx = context.get_admin_context()
+            self.plugin_instance.db.update_loadbalancer_provisioning_status(
+                ctx,
+                loadbalancer['loadbalancer']['id'])
+            new_lb_name = 'new_lb_name'
+            loadbalancer['loadbalancer']['name'] = new_lb_name
+            self._update_loadbalancer_api(
+                lb_id, {'loadbalancer': {'name': new_lb_name}})
+            calls = self.mock_api.update_loadbalancer.call_args_list
+            self.assertEqual(1, len(calls))
+            _, called_old_lb, called_new_lb, called_host = calls[0][0]
+            self.assertEqual(lb_id, called_old_lb.id)
+            self.assertEqual(lb_id, called_new_lb.id)
+            self.assertEqual(old_lb_name, called_old_lb.name)
+            self.assertEqual(new_lb_name, called_new_lb.name)
+            self.assertEqual('host', called_host)
+            self.assertEqual(constants.PENDING_UPDATE,
+                             called_new_lb.provisioning_status)
+
+    def test_delete_loadbalancer(self):
+        with self.loadbalancer(no_delete=True) as loadbalancer:
+            lb_id = loadbalancer['loadbalancer']['id']
+            ctx = context.get_admin_context()
+            self._update_status(models.LoadBalancer, constants.ACTIVE, lb_id)
+            self.plugin_instance.delete_loadbalancer(ctx, lb_id)
+            calls = self.mock_api.delete_loadbalancer.call_args_list
+            self.assertEqual(1, len(calls))
+            _, called_lb, called_host = calls[0][0]
+            self.assertEqual(lb_id, called_lb.id)
+            self.assertEqual('host', called_host)
+            self.assertEqual(constants.PENDING_DELETE,
+                             called_lb.provisioning_status)
+            self.assertRaises(loadbalancerv2.EntityNotFound,
+                              self.plugin_instance.db.get_loadbalancer,
+                              ctx, lb_id)

neutron_lbaas/tests/unit/test_agent_scheduler.py

@@ -0,0 +1,259 @@
+# Copyright (c) 2013 OpenStack Foundation.
+# Copyright 2015 Rackspace
+#
+# 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 copy
+
+import mock
+from neutron.api import extensions
+from neutron.api.v2 import attributes
+from neutron import context
+from neutron.db import agents_db
+from neutron.db import servicetype_db as st_db
+from neutron.extensions import agent
+from neutron import manager
+from neutron.plugins.common import constants as plugin_const
+import neutron.tests.unit.extensions
+from neutron.tests.unit.openvswitch import test_agent_scheduler
+from neutron.tests.unit import test_agent_ext_plugin
+from neutron.tests.unit import test_extensions
+from oslo_config import cfg
+from oslo_utils import timeutils
+from webob import exc
+
+from neutron_lbaas.drivers.haproxy import plugin_driver
+from neutron_lbaas.extensions import lbaas_agentschedulerv2
+from neutron_lbaas.services.loadbalancer import constants as lb_const
+from neutron_lbaas.tests import base
+from neutron_lbaas.tests.unit.db.loadbalancer import test_db_loadbalancerv2
+
+LBAAS_HOSTA = 'hosta'
+extensions_path = ':'.join(neutron.tests.unit.extensions.__path__)
+
+
+class AgentSchedulerTestMixIn(test_agent_scheduler.AgentSchedulerTestMixIn):
+    def _list_loadbalancers_hosted_by_agent(
+            self, agent_id, expected_code=exc.HTTPOk.code, admin_context=True):
+        path = "/agents/%s/%s.%s" % (agent_id,
+                                     lbaas_agentschedulerv2.LOADBALANCERS,
+                                     self.fmt)
+        return self._request_list(path, expected_code=expected_code,
+                                  admin_context=admin_context)
+
+    def _get_lbaas_agent_hosting_loadbalancer(self, loadbalancer_id,
+                                              expected_code=exc.HTTPOk.code,
+                                              admin_context=True):
+        path = "/lbaas/loadbalancers/%s/%s.%s" % (loadbalancer_id,
+                                                  lbaas_agentschedulerv2
+                                                  .LOADBALANCER_AGENT,
+                                                  self.fmt)
+        return self._request_list(path, expected_code=expected_code,
+                                  admin_context=admin_context)
+
+
+class LBaaSAgentSchedulerTestCase(test_agent_ext_plugin.AgentDBTestMixIn,
+                                  AgentSchedulerTestMixIn,
+                                  test_db_loadbalancerv2.LbaasTestMixin,
+                                  base.NeutronDbPluginV2TestCase):
+    fmt = 'json'
+    plugin_str = 'neutron.plugins.ml2.plugin.Ml2Plugin'
+
+    def _register_agent_states(self, lbaas_agents=False):
+        res = super(LBaaSAgentSchedulerTestCase, self)._register_agent_states(
+            lbaas_agents=lbaas_agents)
+        if lbaas_agents:
+            lbaas_hosta = {
+                'binary': 'neutron-loadbalancer-agent',
+                'host': test_agent_ext_plugin.LBAAS_HOSTA,
+                'topic': 'LOADBALANCER_AGENT',
+                'configurations': {'device_drivers': [
+                    plugin_driver.HaproxyOnHostPluginDriver.device_driver]},
+                'agent_type': lb_const.AGENT_TYPE_LOADBALANCERV2}
+            lbaas_hostb = copy.deepcopy(lbaas_hosta)
+            lbaas_hostb['host'] = test_agent_ext_plugin.LBAAS_HOSTB
+            callback = agents_db.AgentExtRpcCallback()
+            callback.report_state(self.adminContext,
+                                  agent_state={'agent_state': lbaas_hosta},
+                                  time=timeutils.strtime())
+            callback.report_state(self.adminContext,
+                                  agent_state={'agent_state': lbaas_hostb},
+                                  time=timeutils.strtime())
+            res += [lbaas_hosta, lbaas_hostb]
+        return res
+
+    def setUp(self):
+        # Save the global RESOURCE_ATTRIBUTE_MAP
+        self.saved_attr_map = {}
+        for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
+            self.saved_attr_map[resource] = attrs.copy()
+        service_plugins = {
+            'lb_plugin_name': test_db_loadbalancerv2.DB_LB_PLUGIN_CLASS}
+
+        # default provider should support agent scheduling
+        cfg.CONF.set_override(
+            'service_provider',
+            [('LOADBALANCERV2:lbaas:neutron_lbaas.drivers.haproxy.'
+              'plugin_driver.HaproxyOnHostPluginDriver:default')],
+            'service_providers')
+
+        # need to reload provider configuration
+        st_db.ServiceTypeManager._instance = None
+
+        super(LBaaSAgentSchedulerTestCase, self).setUp(
+            self.plugin_str, service_plugins=service_plugins)
+        ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
+        self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
+        self.adminContext = context.get_admin_context()
+        # Add the resources to the global attribute map
+        # This is done here as the setup process won't
+        # initialize the main API router which extends
+        # the global attribute map
+        attributes.RESOURCE_ATTRIBUTE_MAP.update(
+            agent.RESOURCE_ATTRIBUTE_MAP)
+        self.lbaas_plugin = manager.NeutronManager.get_service_plugins()[
+            plugin_const.LOADBALANCERV2]
+        self.core_plugin = manager.NeutronManager.get_plugin()
+        self.addCleanup(self.restore_attribute_map)
+
+    def restore_attribute_map(self):
+        # Restore the original RESOURCE_ATTRIBUTE_MAP
+        attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
+
+    def test_report_states(self):
+        self._register_agent_states(lbaas_agents=True)
+        agents = self._list_agents()
+        self.assertEqual(8, len(agents['agents']))
+
+    def test_loadbalancer_scheduling_on_loadbalancer_creation(self):
+        self._register_agent_states(lbaas_agents=True)
+        with self.loadbalancer() as loadbalancer:
+            lbaas_agent = self._get_lbaas_agent_hosting_loadbalancer(
+                loadbalancer['loadbalancer']['id'])
+            self.assertIsNotNone(lbaas_agent)
+            self.assertEqual(lbaas_agent['agent']['agent_type'],
+                             lb_const.AGENT_TYPE_LOADBALANCERV2)
+            loadbalancers = self._list_loadbalancers_hosted_by_agent(
+                lbaas_agent['agent']['id'])
+            self.assertEqual(1, len(loadbalancers['loadbalancers']))
+            self.assertEqual(loadbalancer['loadbalancer'],
+                             loadbalancers['loadbalancers'][0])
+            self.lbaas_plugin.db.update_loadbalancer_provisioning_status(
+                self.adminContext, loadbalancer['loadbalancer']['id']
+            )
+
+    def test_schedule_loadbalancer_with_disabled_agent(self):
+        lbaas_hosta = {
+            'binary': 'neutron-loadbalancer-agent',
+            'host': LBAAS_HOSTA,
+            'topic': 'LOADBALANCER_AGENT',
+            'configurations': {'device_drivers': [
+                plugin_driver.HaproxyOnHostPluginDriver.device_driver
+            ]},
+            'agent_type': lb_const.AGENT_TYPE_LOADBALANCERV2}
+        self._register_one_agent_state(lbaas_hosta)
+        with self.loadbalancer() as loadbalancer:
+            lbaas_agent = self._get_lbaas_agent_hosting_loadbalancer(
+                loadbalancer['loadbalancer']['id'])
+            self.assertIsNotNone(lbaas_agent)
+            self.lbaas_plugin.db.update_loadbalancer_provisioning_status(
+                self.adminContext, loadbalancer['loadbalancer']['id']
+            )
+        agents = self._list_agents()
+        self._disable_agent(agents['agents'][0]['id'])
+        subnet = self.core_plugin.get_subnets(self.adminContext)[0]
+        lb = {
+            'loadbalancer': {
+                'vip_subnet_id': subnet['id'],
+                'provider': 'lbaas',
+                'vip_address': attributes.ATTR_NOT_SPECIFIED,
+                'admin_state_up': True,
+                'tenant_id': self._tenant_id}}
+        self.assertRaises(lbaas_agentschedulerv2.NoEligibleLbaasAgent,
+                          self.lbaas_plugin.create_loadbalancer,
+                          self.adminContext, lb)
+
+    def test_schedule_loadbalancer_with_down_agent(self):
+        lbaas_hosta = {
+            'binary': 'neutron-loadbalancer-agent',
+            'host': LBAAS_HOSTA,
+            'topic': 'LOADBALANCER_AGENT',
+            'configurations': {'device_drivers': [
+                plugin_driver.HaproxyOnHostPluginDriver.device_driver
+            ]},
+            'agent_type': lb_const.AGENT_TYPE_LOADBALANCERV2}
+        self._register_one_agent_state(lbaas_hosta)
+        is_agent_down_str = 'neutron.db.agents_db.AgentDbMixin.is_agent_down'
+        with mock.patch(is_agent_down_str) as mock_is_agent_down:
+            mock_is_agent_down.return_value = False
+            with self.loadbalancer() as loadbalancer:
+                lbaas_agent = self._get_lbaas_agent_hosting_loadbalancer(
+                    loadbalancer['loadbalancer']['id'])
+                self.lbaas_plugin.db.update_loadbalancer_provisioning_status(
+                    self.adminContext, loadbalancer['loadbalancer']['id']
+                )
+            self.assertIsNotNone(lbaas_agent)
+        with mock.patch(is_agent_down_str) as mock_is_agent_down:
+            mock_is_agent_down.return_value = True
+            subnet = self.core_plugin.get_subnets(self.adminContext)[0]
+            lb = {
+                'loadbalancer': {
+                    'vip_subnet_id': subnet['id'],
+                    'provider': 'lbaas',
+                    'vip_address': attributes.ATTR_NOT_SPECIFIED,
+                    'admin_state_up': True,
+                    'tenant_id': self._tenant_id}}
+            self.assertRaises(lbaas_agentschedulerv2.NoEligibleLbaasAgent,
+                              self.lbaas_plugin.create_loadbalancer,
+                              self.adminContext, lb)
+
+    def test_loadbalancer_unscheduling_on_loadbalancer_deletion(self):
+        self._register_agent_states(lbaas_agents=True)
+        with self.loadbalancer(no_delete=True) as loadbalancer:
+            lb_id = loadbalancer['loadbalancer']['id']
+            lbaas_agent = self._get_lbaas_agent_hosting_loadbalancer(lb_id)
+            self.assertIsNotNone(lbaas_agent)
+            self.assertEqual(lbaas_agent['agent']['agent_type'],
+                             lb_const.AGENT_TYPE_LOADBALANCERV2)
+            loadbalancers = self._list_loadbalancers_hosted_by_agent(
+                lbaas_agent['agent']['id'])
+            self.assertEqual(1, len(loadbalancers['loadbalancers']))
+            self.assertEqual(loadbalancer['loadbalancer'],
+                             loadbalancers['loadbalancers'][0])
+
+            self.lbaas_plugin.db.update_loadbalancer_provisioning_status(
+                self.adminContext, lb_id
+            )
+
+            req = self.new_delete_request('loadbalancers', lb_id)
+            res = req.get_response(self.ext_api)
+            self.assertEqual(res.status_int, exc.HTTPNoContent.code)
+            loadbalancers = self._list_loadbalancers_hosted_by_agent(
+                lbaas_agent['agent']['id'])
+            self.assertEqual(0, len(loadbalancers['loadbalancers']))
+
+    def test_loadbalancer_scheduling_non_admin_access(self):
+        self._register_agent_states(lbaas_agents=True)
+        with self.loadbalancer() as loadbalancer:
+            self._get_lbaas_agent_hosting_loadbalancer(
+                loadbalancer['loadbalancer']['id'],
+                expected_code=exc.HTTPForbidden.code,
+                admin_context=False)
+            self._list_loadbalancers_hosted_by_agent(
+                'fake_id',
+                expected_code=exc.HTTPForbidden.code,
+                admin_context=False)
+            self.lbaas_plugin.db.update_loadbalancer_provisioning_status(
+                self.adminContext, loadbalancer['loadbalancer']['id']
+            )

setup.cfg

@@ -42,6 +42,8 @@ device_drivers =
     neutron.services.loadbalancer.drivers.haproxy.plugin_driver.HaproxyOnHostPluginDriver = neutron_lbaas.services.loadbalancer.drivers.haproxy.plugin_driver:HaproxyOnHostPluginDriver
     neutron.services.loadbalancer.drivers.netscaler.netscaler_driver.NetScalerPluginDriver = neutron_lbaas.services.loadbalancer.drivers.netscaler.netscaler_driver:NetScalerPluginDriver
     neutron.services.loadbalancer.drivers.radware.driver.LoadBalancerDriver = neutron_lbaas.services.loadbalancer.drivers.radware.driver:LoadBalancerDriver
+loadbalancer_schedulers =
+    neutron_lbaas.agent_scheduler.ChanceScheduler = neutron_lbaas.agent_scheduler:ChanceScheduler
 pool_schedulers =
     neutron.services.loadbalancer.agent_scheduler.ChanceScheduler = neutron_lbaas.services.loadbalancer.agent_scheduler:ChanceScheduler