Stop using scheduler RPC API magic

If a scheduler RPC message isn't handled directly by a SchedulerManager
method, the __getattr__() fallback passes the message to a driver
method in the form of schedule_${method}() and, if that doesn't exist,
instead calls the schedule() method supplying the topic and method
args.

This is pretty bizarre stuff and we appear to only use it in two cases:

  1) live_migration - this is how the schedule_live_migration()
     method in the driver gets called, but the side-effect is that
     we require the client to pass a topic argument which is never
     used. This would be much more sanely handled with an explicit
     SchedulerManager.live_migration() method.

  2) create_volume - the volume API asks the scheduler to pick a
     target host for create_volume() using this method. This would
     be easily handled with an SchedulerManager.create_volume()
     method.

Change-Id: I1047489d85ac51d8d36fea1c4eb858df638ce349

11 files changed, 84 insertions, 20 deletions

diff --git a/nova/compute/api.py b/nova/compute/api.py
index c39e588f3..e47147522 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -1807,7 +1807,7 @@ class API(base.Base):
         LOG.debug(_("Going to try to live migrate instance"),
                   instance=instance)
         self.scheduler_rpcapi.live_migration(context, block_migration,
-                disk_over_commit, instance, host, topic=FLAGS.compute_topic)
+                disk_over_commit, instance, host)
 
 
 class HostAPI(base.Base):
diff --git a/nova/scheduler/chance.py b/nova/scheduler/chance.py
index 9731a0a07..e74f34054 100644
--- a/nova/scheduler/chance.py
+++ b/nova/scheduler/chance.py
@@ -24,8 +24,11 @@ Chance (Random) Scheduler implementation
 import random
 
 from nova import exception
+from nova import flags
 from nova.scheduler import driver
 
+FLAGS = flags.FLAGS
+
 
 class ChanceScheduler(driver.Scheduler):
     """Implements Scheduler as a random node selector."""
@@ -123,3 +126,9 @@ class ChanceScheduler(driver.Scheduler):
                               filter_properties)
         self.compute_rpcapi.prep_resize(context, image, instance,
                 instance_type, host, reservations)
+
+    def schedule_create_volume(self, context, volume_id, snapshot_id,
+                               reservations):
+        self.schedule(context, FLAGS.volume_topic, 'create_volume',
+                      volume_id=volume_id, snapshot_id=snapshot_id,
+                      reservations=reservations)
diff --git a/nova/scheduler/driver.py b/nova/scheduler/driver.py
index 42e05e903..f599965cd 100644
--- a/nova/scheduler/driver.py
+++ b/nova/scheduler/driver.py
@@ -202,6 +202,11 @@ class Scheduler(object):
         msg = _("Driver must implement schedule_run_instance")
         raise NotImplementedError(msg)
 
+    def schedule_create_volume(self, context, volume_id, snapshot_id,
+                               reservations):
+        msg = _("Driver must implement schedule_create_volune")
+        raise NotImplementedError(msg)
+
     def schedule_live_migration(self, context, dest,
                                 block_migration=False, disk_over_commit=False,
                                 instance=None, instance_id=None):
diff --git a/nova/scheduler/filter_scheduler.py b/nova/scheduler/filter_scheduler.py
index 9eea8dea6..258b66f7e 100644
--- a/nova/scheduler/filter_scheduler.py
+++ b/nova/scheduler/filter_scheduler.py
@@ -51,6 +51,13 @@ class FilterScheduler(driver.Scheduler):
         msg = _("No host selection for %s defined.") % topic
         raise exception.NoValidHost(reason=msg)
 
+    def schedule_create_volume(self, context, volume_id, snapshot_id,
+                               reservations):
+        # NOTE: We're only focused on compute instances right now,
+        # so this method will always raise NoValidHost().
+        msg = _("No host selection for %s defined.") % FLAGS.volume_topic
+        raise exception.NoValidHost(reason=msg)
+
     def schedule_run_instance(self, context, request_spec,
                               admin_password, injected_files,
                               requested_networks, is_first_time,
diff --git a/nova/scheduler/manager.py b/nova/scheduler/manager.py
index e1030121a..36e9a8473 100644
--- a/nova/scheduler/manager.py
+++ b/nova/scheduler/manager.py
@@ -53,7 +53,7 @@ QUOTAS = quota.QUOTAS
 class SchedulerManager(manager.Manager):
     """Chooses a host to run instances on."""
 
-    RPC_API_VERSION = '1.6'
+    RPC_API_VERSION = '1.7'
 
     def __init__(self, scheduler_driver=None, *args, **kwargs):
         if not scheduler_driver:
@@ -66,6 +66,8 @@ class SchedulerManager(manager.Manager):
         # NOTE(russellb) Because of what this is doing, we must be careful
         # when changing the API of the scheduler drivers, as that changes
         # the rpc API as well, and the version should be updated accordingly.
+        # NOTE(markmc): This remains only for backwards compat support
+        # and can be removed when we bump the major version
         return functools.partial(self._schedule, key)
 
     def get_host_list(self, context):
@@ -91,6 +93,30 @@ class SchedulerManager(manager.Manager):
         self.driver.update_service_capabilities(service_name, host,
                 capabilities)
 
+    def create_volume(self, context, volume_id, snapshot_id, reservations):
+        try:
+            self.driver.schedule_create_volume(
+                context, volume_id, snapshot_id, reservations)
+        except Exception as ex:
+            with excutils.save_and_reraise_exception():
+                self._set_vm_state_and_notify('create_volume',
+                                             {'vm_state': vm_states.ERROR},
+                                             context, ex, {})
+
+    def live_migration(self, context, dest,
+                       block_migration=False, disk_over_commit=False,
+                       instance=None, instance_id=None, topic=None):
+        try:
+            return self.driver.schedule_live_migration(
+                context, dest,
+                block_migration, disk_over_commit,
+                instance, instance_id)
+        except Exception as ex:
+            with excutils.save_and_reraise_exception():
+                self._set_vm_state_and_notify('live_migration',
+                                             {'vm_state': vm_states.ERROR},
+                                             context, ex, {})
+
     def _schedule(self, method, context, topic, *args, **kwargs):
         """Tries to call schedule_* method on the driver to retrieve host.
         Falls back to schedule(context, topic) if method doesn't exist.
diff --git a/nova/scheduler/multi.py b/nova/scheduler/multi.py
index c6de3f825..59e1212c0 100644
--- a/nova/scheduler/multi.py
+++ b/nova/scheduler/multi.py
@@ -89,6 +89,9 @@ class MultiScheduler(driver.Scheduler):
     def schedule_prep_resize(self, *args, **kwargs):
         return self.drivers['compute'].schedule_prep_resize(*args, **kwargs)
 
+    def schedule_create_volume(self, *args, **kwargs):
+        return self.drivers['volume'].schedule_create_volume(*args, **kwargs)
+
     def update_service_capabilities(self, service_name, host, capabilities):
         # Multi scheduler is only a holder of sub-schedulers, so
         # pass the capabilities to the schedulers that matter
diff --git a/nova/scheduler/rpcapi.py b/nova/scheduler/rpcapi.py
index 84d0286a2..8ff564a59 100644
--- a/nova/scheduler/rpcapi.py
+++ b/nova/scheduler/rpcapi.py
@@ -41,6 +41,7 @@ class SchedulerAPI(nova.openstack.common.rpc.proxy.RpcProxy):
         1.4 - Remove update_db from prep_resize
         1.5 - Add reservations argument to prep_resize()
         1.6 - Remove reservations argument to run_instance()
+        1.7 - Add create_volume() method, remove topic from live_migration()
     '''
 
     BASE_RPC_API_VERSION = '1.0'
@@ -73,14 +74,21 @@ class SchedulerAPI(nova.openstack.common.rpc.proxy.RpcProxy):
         return self.call(ctxt, self.make_msg('show_host_resources', host=host))
 
     def live_migration(self, ctxt, block_migration, disk_over_commit,
-            instance, dest, topic):
+            instance, dest):
         # NOTE(comstud): Call vs cast so we can get exceptions back, otherwise
         # this call in the scheduler driver doesn't return anything.
         instance_p = jsonutils.to_primitive(instance)
         return self.call(ctxt, self.make_msg('live_migration',
                 block_migration=block_migration,
                 disk_over_commit=disk_over_commit, instance=instance_p,
-                dest=dest, topic=topic), version='1.3')
+                dest=dest), version='1.7')
+
+    def create_volume(self, ctxt, volume_id, snapshot_id, reservations):
+        self.cast(ctxt,
+                  self.make_msg('create_volume',
+                                volume_id=volume_id, snapshot_id=snapshot_id,
+                                reservations=reservations),
+                  version='1.7')
 
     def update_service_capabilities(self, ctxt, service_name, host,
             capabilities):
diff --git a/nova/scheduler/simple.py b/nova/scheduler/simple.py
index f8a1e4154..273bb2cf8 100644
--- a/nova/scheduler/simple.py
+++ b/nova/scheduler/simple.py
@@ -54,7 +54,8 @@ class SimpleScheduler(chance.ChanceScheduler):
         super(SimpleScheduler, self).schedule_run_instance(context,
                 request_spec, reservations, *_args, **_kwargs)
 
-    def schedule_create_volume(self, context, volume_id, *_args, **_kwargs):
+    def schedule_create_volume(self, context, volume_id, snapshot_id,
+                               reservations):
         """Picks a host that is up and has the fewest volumes."""
         deprecated.warn(_('nova-volume functionality is deprecated in Folsom '
                 'and will be removed in Grizzly.  Volumes are now handled '
@@ -72,7 +73,8 @@ class SimpleScheduler(chance.ChanceScheduler):
             if not utils.service_is_up(service):
                 raise exception.WillNotSchedule(host=host)
             driver.cast_to_volume_host(context, host, 'create_volume',
-                    volume_id=volume_id, **_kwargs)
+                    volume_id=volume_id, snapshot_id=snapshot_id,
+                    reservations=reservations)
             return None
 
         results = db.service_get_all_volume_sorted(elevated)
@@ -86,7 +88,8 @@ class SimpleScheduler(chance.ChanceScheduler):
                 raise exception.NoValidHost(reason=msg)
             if utils.service_is_up(service) and not service['disabled']:
                 driver.cast_to_volume_host(context, service['host'],
-                        'create_volume', volume_id=volume_id, **_kwargs)
+                        'create_volume', volume_id=volume_id,
+                        snapshot_id=snapshot_id, reservations=reservations)
                 return None
         msg = _("Is the appropriate service running?")
         raise exception.NoValidHost(reason=msg)
diff --git a/nova/tests/scheduler/test_multi_scheduler.py b/nova/tests/scheduler/test_multi_scheduler.py
index fd79fffb3..6fce5bc35 100644
--- a/nova/tests/scheduler/test_multi_scheduler.py
+++ b/nova/tests/scheduler/test_multi_scheduler.py
@@ -46,9 +46,6 @@ class FakeVolumeScheduler(driver.Scheduler):
         super(FakeVolumeScheduler, self).__init__()
         self.is_update_caps_called = False
 
-    def schedule_create_volume(self, *args, **kwargs):
-        pass
-
     def schedule_create_volumes(self, *args, **kwargs):
         pass
 
@@ -97,7 +94,7 @@ class MultiDriverTestCase(test_scheduler.SchedulerTestCase):
 
         #no compute methods are proxied at this time
         test_methods = {compute_driver: [],
-                        volume_driver: ['create_volume', 'create_volumes']}
+                        volume_driver: ['create_volumes']}
 
         for driver, methods in test_methods.iteritems():
             for method in methods:
diff --git a/nova/tests/scheduler/test_rpcapi.py b/nova/tests/scheduler/test_rpcapi.py
index a2fee64b2..95133622b 100644
--- a/nova/tests/scheduler/test_rpcapi.py
+++ b/nova/tests/scheduler/test_rpcapi.py
@@ -86,10 +86,16 @@ class SchedulerRpcAPITestCase(test.TestCase):
         self._test_scheduler_api('live_migration', rpc_method='call',
                 block_migration='fake_block_migration',
                 disk_over_commit='fake_disk_over_commit',
-                instance='fake_instance', dest='fake_dest', topic='fake_topic',
-                version='1.3')
+                instance='fake_instance', dest='fake_dest',
+                version='1.7')
 
     def test_update_service_capabilities(self):
         self._test_scheduler_api('update_service_capabilities',
                 rpc_method='fanout_cast', service_name='fake_name',
                 host='fake_host', capabilities='fake_capabilities')
+
+    def test_create_volume(self):
+        self._test_scheduler_api('create_volume',
+                rpc_method='cast', volume_id="fake_volume",
+                snapshot_id="fake_snapshots", reservations=list('fake_res'),
+                version='1.7')
diff --git a/nova/volume/api.py b/nova/volume/api.py
index bbb3149b0..3b8c62e11 100644
--- a/nova/volume/api.py
+++ b/nova/volume/api.py
@@ -31,6 +31,7 @@ from nova.openstack.common import rpc
 from nova.openstack.common import timeutils
 import nova.policy
 from nova import quota
+from nova.scheduler import rpcapi as scheduler_rpcapi
 
 volume_host_opt = cfg.BoolOpt('snapshot_same_host',
         default=True,
@@ -72,6 +73,10 @@ def check_policy(context, action, target_obj=None):
 class API(base.Base):
     """API for interacting with the volume manager."""
 
+    def __init__(self, **kwargs):
+        self.scheduler_rpcapi = scheduler_rpcapi.SchedulerAPI()
+        super(API, self).__init__(**kwargs)
+
     def create(self, context, size, name, description, snapshot=None,
                      volume_type=None, metadata=None, availability_zone=None):
         check_policy(context, 'create')
@@ -172,13 +177,8 @@ class API(base.Base):
                       "args": {"volume_id": volume_id,
                                "snapshot_id": snapshot_id}})
         else:
-            rpc.cast(context,
-                     FLAGS.scheduler_topic,
-                     {"method": "create_volume",
-                      "args": {"topic": FLAGS.volume_topic,
-                               "volume_id": volume_id,
-                               "snapshot_id": snapshot_id,
-                               "reservations": reservations}})
+            self.scheduler_rpcapi.create_volume(
+                context, volume_id, snapshot_id, reservations)
 
     @wrap_check_policy
     def delete(self, context, volume):