Split scheduler out of engine service

Change-Id: Ib57f295b0362e4fc17a3ff5bfb539830f6ac1875

devstack/plugin.sh

@@ -144,13 +144,14 @@ function install_mogan_pythonclient {
 
 # start_mogan - Start running processes, including screen
 function start_mogan {
-    if is_service_enabled mogan-api && is_service_enabled mogan-engine ; then
+    if is_service_enabled mogan-api && is_service_enabled mogan-engine && is_service_enabled mogan-scheduler; then
         echo_summary "Installing all mogan services in separate processes"
         run_process mogan-api "${MOGAN_BIN_DIR}/mogan-api --config-file ${MOGAN_CONF_DIR}/mogan.conf"
         if ! wait_for_service ${SERVICE_TIMEOUT} ${MOGAN_SERVICE_PROTOCOL}://${MOGAN_SERVICE_HOST}:${MOGAN_SERVICE_PORT}; then
             die $LINENO "mogan-api did not start"
         fi
         run_process mogan-engine "${MOGAN_BIN_DIR}/mogan-engine --config-file ${MOGAN_CONF_DIR}/mogan.conf"
+        run_process mogan-scheduler "${MOGAN_BIN_DIR}/mogan-scheduler --config-file ${MOGAN_CONF_DIR}/mogan.conf"
     fi
 }
 
@@ -158,7 +159,7 @@ function start_mogan {
 # stop_mogan - Stop running processes
 function stop_mogan {
     # Kill the Mogan screen windows
-    for serv in mogan-api mogan-engine; do
+    for serv in mogan-api mogan-engine mogan-scheduler; do
         stop_process $serv
     done
 }

devstack/settings

@@ -2,8 +2,8 @@
 
 # We have to add Mogan to enabled services for run_process to work
 # Now we just support to run services in separate processes and screens:
-# enable_service mogan mogan-api mogan-engine
+# enable_service mogan mogan-api mogan-engine mogan-scheduler
-enable_service mogan mogan-api mogan-engine
+enable_service mogan mogan-api mogan-engine mogan-scheduler
 
 # Set up default repos
 MOGAN_REPO=${MOGAN_REPO:-${GIT_BASE}/openstack/mogan.git}

mogan/cmd/engine.py

@@ -32,10 +32,9 @@ def main():
     # Parse config file and command line options, then start logging
     mogan_service.prepare_service(sys.argv)
 
-    mgr = mogan_service.RPCService(CONF.host,
+    mgr = mogan_service.RPCService('mogan.engine.manager',
-                                   'mogan.engine.manager',
                                    'EngineManager',
-                                   constants.MANAGER_TOPIC)
+                                   constants.ENGINE_TOPIC)
 
     launcher = service.launch(CONF, mgr)
     launcher.wait()

mogan/cmd/scheduler.py

@@ -0,0 +1,40 @@
+# Copyright 2017 Huawei Technologies Co.,LTD.
+# 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.
+
+"""
+The Mogan Scheduler Service
+"""
+
+import sys
+
+from oslo_config import cfg
+from oslo_service import service
+
+from mogan.common import constants
+from mogan.common import service as mogan_service
+
+CONF = cfg.CONF
+
+
+def main():
+    # Parse config file and command line options, then start logging
+    mogan_service.prepare_service(sys.argv)
+
+    mgr = mogan_service.RPCService('mogan.scheduler.manager',
+                                   'SchedulerManager',
+                                   constants.SCHEDULER_TOPIC)
+
+    launcher = service.launch(CONF, mgr)
+    launcher.wait()

mogan/common/constants.py

@@ -14,4 +14,5 @@
 #    under the License.
 
 
-MANAGER_TOPIC = 'mogan.engine_manager'
+ENGINE_TOPIC = 'mogan-engine'
+SCHEDULER_TOPIC = 'mogan-scheduler'

mogan/common/service.py

@@ -36,9 +36,9 @@ LOG = log.getLogger(__name__)
 
 class RPCService(service.Service):
 
-    def __init__(self, host, manager_module, manager_class, topic):
+    def __init__(self, manager_module, manager_class, topic, host=None):
         super(RPCService, self).__init__()
-        self.host = host
+        self.host = host or CONF.host
         manager_module = importutils.try_import(manager_module)
         manager_class = getattr(manager_module, manager_class)
         self.manager = manager_class(host, topic)

mogan/conf/engine.py

@@ -34,10 +34,6 @@ opts = [
                default=60,
                help=_('Interval between syncing the resources from underlying '
                       'hypervisor, in seconds.')),
-    cfg.StrOpt('scheduler_driver',
-               default='mogan.engine.scheduler.filter_scheduler.'
-                       'FilterScheduler',
-               help=_('Default scheduler driver to use')),
     cfg.StrOpt('default_schedule_zone',
                help=_("Availability zone to use for scheduling when user "
                       "doesn't specify one.")),

mogan/conf/scheduler.py

@@ -19,11 +19,10 @@ from mogan.common.i18n import _
 
 opts = [
     cfg.StrOpt('scheduler_driver',
-               default='mogan.engine.scheduler.filter_scheduler.'
+               default='mogan.scheduler.filter_scheduler.FilterScheduler',
-                       'FilterScheduler',
                help=_('Default scheduler driver to use')),
     cfg.StrOpt('scheduler_node_manager',
-               default='mogan.engine.scheduler.node_manager.NodeManager',
+               default='mogan.scheduler.node_manager.NodeManager',
                help=_('The scheduler node manager class to use')),
     cfg.IntOpt('scheduler_max_attempts',
                default=3,
@@ -47,7 +46,7 @@ opts = [
                 help=_('Which weigher class names to use for weighing '
                        'nodes.')),
     cfg.StrOpt('scheduler_weight_handler',
-               default='mogan.engine.scheduler.weights.'
+               default='mogan.scheduler.weights.'
                        'OrderedNodeWeightHandler',
                help=_('Which handler to use for selecting the node after '
                       'weighing')),

mogan/engine/base_manager.py

@@ -17,7 +17,6 @@
 
 from eventlet import greenpool
 from oslo_service import periodic_task
-from oslo_utils import importutils
 
 from mogan.common.i18n import _
 from mogan.conf import CONF
@@ -25,6 +24,7 @@ from mogan.db import api as dbapi
 from mogan.engine.baremetal import driver
 from mogan.engine import rpcapi
 from mogan import network
+from mogan.scheduler import rpcapi as scheduler_rpcapi
 
 
 class BaseEngineManager(periodic_task.PeriodicTasks):
@@ -36,8 +36,7 @@ class BaseEngineManager(periodic_task.PeriodicTasks):
         self.host = host
         self.topic = topic
         self.network_api = network.API()
-        scheduler_driver = CONF.scheduler.scheduler_driver
+        self.scheduler_rpcapi = scheduler_rpcapi.SchedulerAPI()
-        self.scheduler = importutils.import_object(scheduler_driver)
         self.driver = driver.load_engine_driver(CONF.engine.engine_driver)
         self.engine_rpcapi = rpcapi.EngineAPI()
         self._sync_power_pool = greenpool.GreenPool(

mogan/engine/flows/create_instance.py

@@ -47,11 +47,10 @@ class ScheduleCreateInstanceTask(flow_utils.MoganTask):
         self.manager = manager
 
     def execute(self, context, instance, request_spec, filter_properties):
-        with self.manager._lock:
+        top_node = self.manager.scheduler_rpcapi.select_destinations(
-            top_node = self.manager.scheduler.schedule(
+            context,
-                context,
+            request_spec,
-                request_spec,
+            filter_properties)
-                filter_properties)
         instance.node_uuid = top_node
         instance.save()
 

mogan/engine/manager.py

@@ -13,8 +13,6 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import threading
-
 from oslo_log import log
 import oslo_messaging as messaging
 from oslo_service import periodic_task
@@ -45,8 +43,6 @@ class EngineManager(base_manager.BaseEngineManager):
     RPC_API_VERSION = '1.0'
 
     target = messaging.Target(version=RPC_API_VERSION)
-    # TODO(zhenguo): Move lock to scheduler
-    _lock = threading.Lock()
 
     def _get_compute_port(self, context, port_uuid):
         """Gets compute port by the uuid."""

mogan/engine/rpcapi.py

@@ -40,7 +40,7 @@ class EngineAPI(object):
         super(EngineAPI, self).__init__()
         self.topic = topic
         if self.topic is None:
-            self.topic = constants.MANAGER_TOPIC
+            self.topic = constants.ENGINE_TOPIC
 
         target = messaging.Target(topic=self.topic,
                                   version='1.0')

mogan/scheduler/base_filter.py

@@ -20,7 +20,7 @@ from oslo_log import log as logging
 import six
 
 from mogan.common.i18n import _LI
-from mogan.engine.scheduler import base_handler
+from mogan.scheduler import base_handler
 
 LOG = logging.getLogger(__name__)
 

mogan/scheduler/base_weight.py

@@ -21,7 +21,7 @@ import abc
 
 import six
 
-from mogan.engine.scheduler import base_handler
+from mogan.scheduler import base_handler
 
 
 def normalize(weight_list, minval=None, maxval=None):

mogan/scheduler/filter_scheduler.py

@@ -24,8 +24,9 @@ from mogan.common import exception
 from mogan.common.i18n import _
 from mogan.common.i18n import _LE
 from mogan.common.i18n import _LW
-from mogan.engine.scheduler import driver
+from mogan.common import utils
-from mogan.engine.scheduler import scheduler_options
+from mogan.scheduler import driver
+from mogan.scheduler import scheduler_options
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -170,18 +171,28 @@ class FilterScheduler(driver.Scheduler):
         return weighed_nodes
 
     def schedule(self, context, request_spec, filter_properties=None):
-        weighed_nodes = self._get_weighted_candidates(context, request_spec,
-                                                      filter_properties)
-        if not weighed_nodes:
-            LOG.warning(_LW('No weighed nodes found for instance '
-                            'with properties: %s'),
-                        request_spec.get('instance_type'))
-            raise exception.NoValidNode(_("No weighed nodes available"))
 
-        top_node = self._choose_top_node(weighed_nodes, request_spec)
+        # TODO(zhenguo): Scheduler API is inherently multi-threaded as every
-        top_node.obj.consume_from_request(context)
+        # incoming RPC message will be dispatched in it's own green thread.
-        self._add_retry_node(filter_properties, top_node.obj.node)
+        # So we add a syncronized here to make sure the shared node states
-        return top_node.obj.node
+        # consistent, but lock the whole schedule process is not a good choice,
+        # we need to improve this.
+        @utils.synchronized('schedule')
+        def _schedule(self, context, request_spec, filter_properties):
+            weighed_nodes = self._get_weighted_candidates(
+                context, request_spec, filter_properties)
+            if not weighed_nodes:
+                LOG.warning(_LW('No weighed nodes found for instance '
+                                'with properties: %s'),
+                            request_spec.get('instance_type'))
+                raise exception.NoValidNode(_("No weighed nodes available"))
+
+            top_node = self._choose_top_node(weighed_nodes, request_spec)
+            top_node.obj.consume_from_request(context)
+            self._add_retry_node(filter_properties, top_node.obj.node)
+            return top_node.obj.node
+
+        return _schedule(self, context, request_spec, filter_properties)
 
     def _choose_top_node(self, weighed_nodes, request_spec):
         return weighed_nodes[0]

mogan/scheduler/filters/__init__.py

@@ -17,7 +17,7 @@
 Scheduler node filters
 """
 
-from mogan.engine.scheduler import base_filter
+from mogan.scheduler import base_filter
 
 
 class BaseNodeFilter(base_filter.BaseFilter):

mogan/scheduler/filters/availability_zone_filter.py

@@ -13,7 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from mogan.engine.scheduler import filters
+from mogan.scheduler import filters
 
 
 class AvailabilityZoneFilter(filters.BaseNodeFilter):

mogan/scheduler/filters/capabilities_filter.py

@@ -15,8 +15,8 @@
 
 from oslo_log import log as logging
 
-from mogan.engine.scheduler import filters
+from mogan.scheduler import filters
-from mogan.engine.scheduler.filters import extra_specs_ops
+from mogan.scheduler.filters import extra_specs_ops
 
 LOG = logging.getLogger(__name__)
 

mogan/scheduler/filters/instance_type_filter.py

@@ -13,7 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-from mogan.engine.scheduler import filters
+from mogan.scheduler import filters
 
 
 class InstanceTypeFilter(filters.BaseNodeFilter):

mogan/scheduler/filters/json_filter.py

@@ -18,7 +18,7 @@ import operator
 from oslo_serialization import jsonutils
 import six
 
-from mogan.engine.scheduler import filters
+from mogan.scheduler import filters
 
 
 class JsonFilter(filters.BaseNodeFilter):

mogan/scheduler/filters/ports_filter.py

@@ -15,7 +15,7 @@
 
 from oslo_log import log as logging
 
-from mogan.engine.scheduler import filters
+from mogan.scheduler import filters
 
 LOG = logging.getLogger(__name__)
 

mogan/scheduler/manager.py

@@ -0,0 +1,58 @@
+# Copyright 2017 Huawei Technologies Co.,LTD.
+# 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 eventlet
+import oslo_messaging as messaging
+from oslo_service import periodic_task
+from oslo_utils import importutils
+
+from mogan.common import exception
+from mogan.conf import CONF
+
+
+class SchedulerManager(periodic_task.PeriodicTasks):
+    """Mogan Scheduler manager main class."""
+
+    RPC_API_VERSION = '1.0'
+
+    target = messaging.Target(version=RPC_API_VERSION)
+
+    def __init__(self, topic, host=None):
+        super(SchedulerManager, self).__init__(CONF)
+        self.host = host or CONF.host
+        self.topic = topic
+        scheduler_driver = CONF.scheduler.scheduler_driver
+        self.driver = importutils.import_object(scheduler_driver)
+        self._startup_delay = True
+
+    def init_host(self):
+        self._startup_delay = False
+
+    def _wait_for_scheduler(self):
+        while self._startup_delay and not self.driver.is_ready():
+            eventlet.sleep(1)
+
+    @messaging.expected_exceptions(exception.NoValidNode)
+    def select_destinations(self, ctxt, request_spec, filter_properties):
+        self._wait_for_scheduler()
+        dests = self.driver.schedule(
+            ctxt, request_spec, filter_properties)
+        return dests
+
+    def del_host(self):
+        pass
+
+    def periodic_tasks(self, context, raise_on_error=False):
+        return self.run_periodic_tasks(context, raise_on_error=raise_on_error)

mogan/scheduler/node_manager.py

@@ -22,8 +22,8 @@ from oslo_log import log as logging
 from oslo_utils import importutils
 
 from mogan.common import exception
-from mogan.engine.scheduler import filters
 from mogan import objects
+from mogan.scheduler import filters
 
 
 CONF = cfg.CONF
@@ -52,12 +52,12 @@ class NodeManager(object):
     node_state_cls = NodeState
 
     def __init__(self):
-        self.filter_handler = filters.NodeFilterHandler('mogan.engine.'
+        self.filter_handler = filters.NodeFilterHandler(
-                                                        'scheduler.filters')
+            'mogan.scheduler.filters')
         self.filter_classes = self.filter_handler.get_all_classes()
         self.weight_handler = importutils.import_object(
             CONF.scheduler.scheduler_weight_handler,
-            'mogan.engine.scheduler.weights')
+            'mogan.scheduler.weights')
         self.weight_classes = self.weight_handler.get_all_classes()
 
     def _choose_node_filters(self, filter_cls_names):

mogan/scheduler/rpcapi.py

@@ -0,0 +1,58 @@
+# Copyright 2017 Huawei Technologies Co.,LTD.
+# 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.
+
+"""
+Client side of the scheduler manager RPC API.
+"""
+
+from oslo_config import cfg
+import oslo_messaging as messaging
+
+from mogan.common import constants
+from mogan.common import rpc
+from mogan.objects import base as objects_base
+
+CONF = cfg.CONF
+
+
+class SchedulerAPI(object):
+    """Client side of the scheduler RPC API.
+
+    API version history:
+
+    |    1.0 - Initial version.
+
+    """
+
+    RPC_API_VERSION = '1.0'
+
+    def __init__(self, topic=None):
+        super(SchedulerAPI, self).__init__()
+        self.topic = topic
+        if self.topic is None:
+            self.topic = constants.SCHEDULER_TOPIC
+
+        target = messaging.Target(topic=self.topic,
+                                  version='1.0')
+        serializer = objects_base.MoganObjectSerializer()
+        self.client = rpc.get_client(target,
+                                     version_cap=self.RPC_API_VERSION,
+                                     serializer=serializer)
+
+    def select_destinations(self, context, request_spec, filter_properties):
+        cctxt = self.client.prepare(topic=self.topic, server=CONF.host)
+        return cctxt.call(context, 'select_destinations',
+                          request_spec=request_spec,
+                          filter_properties=filter_properties)

mogan/scheduler/weights/__init__.py

@@ -17,7 +17,7 @@
 Scheduler node weights
 """
 
-from mogan.engine.scheduler import base_weight
+from mogan.scheduler import base_weight
 
 
 class WeighedNode(base_weight.WeighedObject):

mogan/scheduler/weights/port.py

@@ -22,7 +22,7 @@ to a positive number and the weighing has the opposite effect of the default.
 
 from oslo_config import cfg
 
-from mogan.engine.scheduler import weights
+from mogan.scheduler import weights
 
 CONF = cfg.CONF
 

mogan/tests/unit/common/test_service.py

@@ -32,11 +32,10 @@ class TestRPCService(base.TestCase):
 
     def setUp(self):
         super(TestRPCService, self).setUp()
-        host = "fake_host"
         mgr_module = "mogan.engine.manager"
         mgr_class = "EngineManager"
-        self.rpc_svc = service.RPCService(host, mgr_module, mgr_class,
+        self.rpc_svc = service.RPCService(mgr_module, mgr_class,
-                                          constants.MANAGER_TOPIC)
+                                          constants.ENGINE_TOPIC)
 
     @mock.patch.object(oslo_messaging, 'Target', autospec=True)
     @mock.patch.object(objects_base, 'MoganObjectSerializer', autospec=True)
@@ -47,7 +46,7 @@ class TestRPCService(base.TestCase):
         self.rpc_svc.handle_signal = mock.MagicMock()
         self.rpc_svc.start()
         mock_target.assert_called_once_with(topic=self.rpc_svc.topic,
-                                            server="fake_host")
+                                            server="fake-mini")
         mock_ios.assert_called_once_with()
         mock_init_method.assert_called_once_with(self.rpc_svc.manager)
 

mogan/tests/unit/engine/flows/test_create_instance_flow.py

@@ -21,8 +21,8 @@ from oslo_utils import uuidutils
 from mogan.engine.baremetal.ironic import IronicDriver
 from mogan.engine.flows import create_instance
 from mogan.engine import manager
-from mogan.engine.scheduler import filter_scheduler as scheduler
 from mogan import objects
+from mogan.scheduler import rpcapi as scheduler_rpcapi
 from mogan.tests import base
 from mogan.tests.unit.objects import utils as obj_utils
 
@@ -34,13 +34,14 @@ class CreateInstanceFlowTestCase(base.TestCase):
         self.ctxt = context.get_admin_context()
 
     @mock.patch.object(objects.instance.Instance, 'save')
-    @mock.patch.object(scheduler.FilterScheduler, 'schedule')
+    @mock.patch.object(scheduler_rpcapi.SchedulerAPI, 'select_destinations')
     def test_schedule_task_execute(self, mock_schedule, mock_save):
         fake_uuid = uuidutils.generate_uuid()
         fake_engine_manager = mock.MagicMock()
+        sche_rpcapi = scheduler_rpcapi.SchedulerAPI()
+        fake_engine_manager.scheduler_rpcapi = sche_rpcapi
         fake_request_spec = mock.MagicMock()
         fake_filter_props = mock.MagicMock()
-        fake_engine_manager.scheduler = scheduler.FilterScheduler()
         task = create_instance.ScheduleCreateInstanceTask(
             fake_engine_manager)
         instance_obj = obj_utils.get_test_instance(self.ctxt)

mogan/tests/unit/scheduler/fakes.py

@@ -17,8 +17,8 @@
 Fakes For Scheduler tests.
 """
 
-from mogan.engine.scheduler import filter_scheduler
+from mogan.scheduler import filter_scheduler
-from mogan.engine.scheduler import node_manager
+from mogan.scheduler import node_manager
 
 
 class FakeFilterScheduler(filter_scheduler.FilterScheduler):

mogan/tests/unit/scheduler/test_base_filter.py

@@ -15,7 +15,7 @@
 
 import mock
 
-from mogan.engine.scheduler import base_filter
+from mogan.scheduler import base_filter
 from mogan.tests import base as test
 
 

mogan/tests/unit/scheduler/test_node_manager.py

@@ -21,10 +21,10 @@ from oslo_context import context
 from oslo_versionedobjects import base as object_base
 
 from mogan.common import exception
-from mogan.engine.scheduler import filters
-from mogan.engine.scheduler import node_manager
-from mogan.engine.scheduler.node_manager import NodeState
 from mogan.objects import compute_port
+from mogan.scheduler import filters
+from mogan.scheduler import node_manager
+from mogan.scheduler.node_manager import NodeState
 from mogan.tests import base as test
 from mogan.tests.unit.objects import utils as obj_utils
 
@@ -73,7 +73,7 @@ class NodeManagerTestCase(test.TestCase):
         self.assertEqual(1, len(filter_classes))
         self.assertEqual('FakeFilterClass2', filter_classes[0].__name__)
 
-    @mock.patch('mogan.engine.scheduler.node_manager.NodeManager.'
+    @mock.patch('mogan.scheduler.node_manager.NodeManager.'
                 '_choose_node_filters')
     def test_get_filtered_nodes(self, _mock_choose_node_filters):
         filter_class = FakeFilterClass1

mogan/tests/unit/scheduler/test_rpcapi.py

@@ -0,0 +1,94 @@
+#
+#    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.
+"""
+Unit Tests for :py:class:`mogan.scheduler.rpcapi.SchedulerAPI`.
+"""
+
+import copy
+
+import mock
+from oslo_config import cfg
+from oslo_messaging import _utils as messaging_utils
+
+from mogan.scheduler import manager as scheduler_manager
+from mogan.scheduler import rpcapi as scheduler_rpcapi
+from mogan.tests import base as tests_base
+from mogan.tests.unit.db import base
+
+CONF = cfg.CONF
+
+
+class SchedulerRPCAPITestCase(tests_base.TestCase):
+
+    def test_versions_in_sync(self):
+        self.assertEqual(
+            scheduler_manager.SchedulerManager.RPC_API_VERSION,
+            scheduler_rpcapi.SchedulerAPI.RPC_API_VERSION)
+
+
+class RPCAPITestCase(base.DbTestCase):
+
+    def _test_rpcapi(self, method, rpc_method, **kwargs):
+        rpcapi = scheduler_rpcapi.SchedulerAPI(topic='fake-topic')
+
+        expected_retval = 'hello world' if rpc_method == 'call' else None
+
+        expected_topic = 'fake-topic'
+
+        target = {
+            "topic": expected_topic,
+            "server": CONF.host,
+            "version": kwargs.pop('version', rpcapi.RPC_API_VERSION)
+        }
+        expected_msg = copy.deepcopy(kwargs)
+
+        self.fake_args = None
+        self.fake_kwargs = None
+
+        def _fake_can_send_version_method(version):
+            return messaging_utils.version_is_compatible(
+                rpcapi.RPC_API_VERSION, version)
+
+        def _fake_prepare_method(*args, **kwargs):
+            for kwd in kwargs:
+                self.assertEqual(kwargs[kwd], target[kwd])
+            return rpcapi.client
+
+        def _fake_rpc_method(*args, **kwargs):
+            self.fake_args = args
+            self.fake_kwargs = kwargs
+            if expected_retval:
+                return expected_retval
+
+        with mock.patch.object(rpcapi.client,
+                               "can_send_version") as mock_can_send_version:
+            mock_can_send_version.side_effect = _fake_can_send_version_method
+            with mock.patch.object(rpcapi.client, "prepare") as mock_prepared:
+                mock_prepared.side_effect = _fake_prepare_method
+
+                with mock.patch.object(rpcapi.client,
+                                       rpc_method) as mock_method:
+                    mock_method.side_effect = _fake_rpc_method
+                    retval = getattr(rpcapi, method)(self.context, **kwargs)
+                    self.assertEqual(retval, expected_retval)
+                    expected_args = [self.context, method, expected_msg]
+                    for arg, expected_arg in zip(self.fake_args,
+                                                 expected_args):
+                        self.assertEqual(arg, expected_arg)
+
+    def test_select_destinations(self):
+        self._test_rpcapi('select_destinations',
+                          'call',
+                          version='1.0',
+                          request_spec=None,
+                          filter_properties=None)

mogan/tests/unit/scheduler/test_scheduler_options.py

@@ -21,7 +21,7 @@ import datetime
 from oslo_serialization import jsonutils
 import six
 
-from mogan.engine.scheduler import scheduler_options
+from mogan.scheduler import scheduler_options
 from mogan.tests import base as test
 
 

mogan/tests/unit/scheduler/test_weights.py

@@ -17,7 +17,7 @@
 Tests For Scheduler weights.
 """
 
-from mogan.engine.scheduler import base_weight
+from mogan.scheduler import base_weight
 from mogan.tests import base as test
 
 

setup.cfg

@@ -24,14 +24,14 @@ packages =
     mogan
 
 [entry_points]
-mogan.engine.scheduler.filters =
+mogan.scheduler.filters =
-    AvailabilityZoneFilter = mogan.engine.scheduler.filters.availability_zone_filter:AvailabilityZoneFilter
+    AvailabilityZoneFilter = mogan.scheduler.filters.availability_zone_filter:AvailabilityZoneFilter
-    InstanceTypeFilter = mogan.engine.scheduler.filters.instance_type_filter:InstanceTypeFilter
+    InstanceTypeFilter = mogan.scheduler.filters.instance_type_filter:InstanceTypeFilter
-    CapabilitiesFilter = mogan.engine.scheduler.filters.capabilities_filter:CapabilitiesFilter
+    CapabilitiesFilter = mogan.scheduler.filters.capabilities_filter:CapabilitiesFilter
-    PortsFilter = mogan.engine.scheduler.filters.ports_filter:PortsFilter
+    PortsFilter = mogan.scheduler.filters.ports_filter:PortsFilter
-    JsonFilter = mogan.engine.scheduler.filters.json_filter:JsonFilter
+    JsonFilter = mogan.scheduler.filters.json_filter:JsonFilter
-mogan.engine.scheduler.weights =
+mogan.scheduler.weights =
-    PortWeigher = mogan.engine.scheduler.weights.port:PortWeigher
+    PortWeigher = mogan.scheduler.weights.port:PortWeigher
 
 oslo.config.opts =
     mogan = mogan.conf.opts:list_opts
@@ -41,8 +41,9 @@ oslo.policy.policies =
 
 console_scripts =
     mogan-api = mogan.cmd.api:main
-    mogan-engine = mogan.cmd.engine:main
     mogan-dbsync = mogan.cmd.dbsync:main
+    mogan-engine = mogan.cmd.engine:main
+    mogan-scheduler = mogan.cmd.scheduler:main
 
 mogan.database.migration_backend =
     sqlalchemy = mogan.db.sqlalchemy.migration