compute.api call conductor ComputeTaskManager for live-migrate

Before the compute api called the scheduler to perform
live migrate. This change makes the compute api call the
conductor, which in turn calls the scheduler.

For those not using the conductor service,
this does not add any extra rpc calls.

This change in control flow means we can look to move
live-migrate more related rpc calls into the conductor
from both schduler driver and compute manager.

part of blueprint live-migration-to-conductor
Change-Id: I081c5e090921d69878d044839616a48f198ecb7d

6 files changed, 156 insertions, 40 deletions

diff --git a/nova/compute/api.py b/nova/compute/api.py
index 19c59bbd5..927476810 100644
--- a/nova/compute/api.py
+++ b/nova/compute/api.py
@@ -197,10 +197,20 @@ class API(base.Base):
         self.consoleauth_rpcapi = consoleauth_rpcapi.ConsoleAuthAPI()
         self.scheduler_rpcapi = scheduler_rpcapi.SchedulerAPI()
         self.compute_rpcapi = compute_rpcapi.ComputeAPI()
+        self._compute_task_api = None
         self.servicegroup_api = servicegroup.API()
 
         super(API, self).__init__(**kwargs)
 
+    @property
+    def compute_task_api(self):
+        if self._compute_task_api is None:
+            # TODO(alaski): Remove calls into here from conductor manager so
+            # that this isn't necessary. #1180540
+            from nova import conductor
+            self._compute_task_api = conductor.ComputeTaskAPI()
+        return self._compute_task_api
+
     def _instance_update(self, context, instance_uuid, **kwargs):
         """Update an instance in the database using kwargs as value."""
 
@@ -2501,8 +2511,11 @@ class API(base.Base):
                                task_state=task_states.MIGRATING,
                                expected_task_state=None)
 
-        self.scheduler_rpcapi.live_migration(context, block_migration,
-                disk_over_commit, instance, host_name)
+        self.compute_task_api.migrate_server(context, instance,
+                scheduler_hint={'host': host_name},
+                live=True, rebuild=False, flavor=None,
+                block_migration=block_migration,
+                disk_over_commit=disk_over_commit)
 
     @check_instance_state(vm_state=[vm_states.ACTIVE, vm_states.STOPPED],
                           task_state=[None])
diff --git a/nova/conductor/api.py b/nova/conductor/api.py
index 63ad11064..ec247d1ac 100644
--- a/nova/conductor/api.py
+++ b/nova/conductor/api.py
@@ -344,6 +344,11 @@ class LocalComputeTaskAPI(object):
         self._manager = utils.ExceptionHelper(
                 manager.ComputeTaskManager())
 
+    def migrate_server(self, context, instance, scheduler_hint, live, rebuild,
+                  flavor, block_migration, disk_over_commit):
+        return self._manager.migrate_server(context, instance, scheduler_hint,
+            live, rebuild, flavor, block_migration, disk_over_commit)
+
 
 class API(object):
     """Conductor API that does updates via RPC to the ConductorManager."""
@@ -681,3 +686,9 @@ class ComputeTaskAPI(object):
 
     def __init__(self):
         self.conductor_compute_rpcapi = rpcapi.ComputeTaskAPI()
+
+    def migrate_server(self, context, instance, scheduler_hint, live, rebuild,
+                  flavor, block_migration, disk_over_commit):
+        return self.conductor_compute_rpcapi.migrate_server(context, instance,
+            scheduler_hint, live, rebuild, flavor, block_migration,
+            disk_over_commit)
diff --git a/nova/conductor/manager.py b/nova/conductor/manager.py
index 0abf8e380..c7f139e4f 100644
--- a/nova/conductor/manager.py
+++ b/nova/conductor/manager.py
@@ -32,6 +32,7 @@ from nova.openstack.common.notifier import api as notifier
 from nova.openstack.common.rpc import common as rpc_common
 from nova.openstack.common import timeutils
 from nova import quota
+from nova.scheduler import rpcapi as scheduler_rpcapi
 
 LOG = logging.getLogger(__name__)
 
@@ -513,7 +514,24 @@ class ComputeTaskManager(object):
     """
 
     RPC_API_NAMESPACE = 'compute_task'
-    RPC_API_VERSION = '1.0'
+    RPC_API_VERSION = '1.1'
 
     def __init__(self):
-        pass
+        self.scheduler_rpcapi = scheduler_rpcapi.SchedulerAPI()
+
+    @rpc_common.client_exceptions(exception.NoValidHost,
+                                  exception.ComputeServiceUnavailable,
+                                  exception.InvalidHypervisorType,
+                                  exception.UnableToMigrateToSelf,
+                                  exception.DestinationHypervisorTooOld,
+                                  exception.InvalidLocalStorage,
+                                  exception.InvalidSharedStorage,
+                                  exception.MigrationPreCheckError)
+    def migrate_server(self, context, instance, scheduler_hint, live, rebuild,
+                  flavor, block_migration, disk_over_commit):
+        if not live or rebuild or (flavor != None):
+            raise NotImplementedError()
+
+        destination = scheduler_hint.get("host")
+        self.scheduler_rpcapi.live_migration(context, block_migration,
+                disk_over_commit, instance, destination)
diff --git a/nova/conductor/rpcapi.py b/nova/conductor/rpcapi.py
index c8d3862d9..c2a0e1b93 100644
--- a/nova/conductor/rpcapi.py
+++ b/nova/conductor/rpcapi.py
@@ -461,6 +461,7 @@ class ComputeTaskAPI(nova.openstack.common.rpc.proxy.RpcProxy):
     API version history:
 
     1.0 - Initial version (empty).
+    1.1 - Added unified migrate_server call.
     """
 
     BASE_RPC_API_VERSION = '1.0'
@@ -470,3 +471,13 @@ class ComputeTaskAPI(nova.openstack.common.rpc.proxy.RpcProxy):
         super(ComputeTaskAPI, self).__init__(
                 topic=CONF.conductor.topic,
                 default_version=self.BASE_RPC_API_VERSION)
+
+    def migrate_server(self, context, instance, scheduler_hint, live, rebuild,
+                  flavor, block_migration, disk_over_commit):
+        instance_p = jsonutils.to_primitive(instance)
+        flavor_p = jsonutils.to_primitive(flavor)
+        msg = self.make_msg('migrate_server', instance=instance_p,
+            scheduler_hint=scheduler_hint, live=live, rebuild=rebuild,
+            flavor=flavor_p, block_migration=block_migration,
+            disk_over_commit=disk_over_commit)
+        return self.call(context, msg, version='1.1')
diff --git a/nova/tests/api/openstack/compute/contrib/test_admin_actions.py b/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
index 2efb6fe5a..5e64bdf94 100644
--- a/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
+++ b/nova/tests/api/openstack/compute/contrib/test_admin_actions.py
@@ -22,10 +22,10 @@ from nova.api.openstack import compute
 from nova.api.openstack.compute.contrib import admin_actions
 from nova.compute import api as compute_api
 from nova.compute import vm_states
+from nova.conductor import api as conductor_api
 from nova import context
 from nova import exception
 from nova.openstack.common import jsonutils
-from nova.scheduler import rpcapi as scheduler_rpcapi
 from nova import test
 from nova.tests.api.openstack import fakes
 
@@ -140,16 +140,15 @@ class AdminActionsTest(test.TestCase):
                         task_state, expected_task_state):
             return None
 
-        def fake_scheduler_api_live_migration(self, context, dest,
-                                        block_migration=False,
-                                        disk_over_commit=False, instance=None,
-                                        instance_id=None, topic=None):
+        def fake_migrate_server(self, context, instance,
+                scheduler_hint, live, rebuild, flavor,
+                block_migration, disk_over_commit):
             return None
 
         self.stubs.Set(compute_api.API, 'update', fake_update)
-        self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
-                       'live_migration',
-                       fake_scheduler_api_live_migration)
+        self.stubs.Set(conductor_api.ComputeTaskAPI,
+                       'migrate_server',
+                       fake_migrate_server)
 
         res = req.get_response(app)
         self.assertEqual(res.status_int, 202)
@@ -194,16 +193,15 @@ class AdminActionsTest(test.TestCase):
                         task_state, expected_task_state):
             return None
 
-        def fake_scheduler_api_live_migration(context, dest,
-                                        block_migration=False,
-                                        disk_over_commit=False, instance=None,
-                                        instance_id=None, topic=None):
+        def fake_migrate_server(self, context, instance,
+                scheduler_hint, live, rebuild, flavor,
+                block_migration, disk_over_commit):
             raise exception.ComputeServiceUnavailable(host='host')
 
         self.stubs.Set(compute_api.API, 'update', fake_update)
-        self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
-                       'live_migration',
-                       fake_scheduler_api_live_migration)
+        self.stubs.Set(conductor_api.ComputeTaskAPI,
+                       'migrate_server',
+                       fake_migrate_server)
 
         res = req.get_response(app)
         self.assertEqual(res.status_int, 400)
@@ -232,16 +230,15 @@ class AdminActionsTest(test.TestCase):
                         task_state, expected_task_state):
             return None
 
-        def fake_scheduler_api_live_migration(context, dest,
-                                        block_migration=False,
-                                        disk_over_commit=False, instance=None,
-                                        instance_id=None, topic=None):
+        def fake_migrate_server(self, context, instance,
+                scheduler_hint, live, rebuild, flavor,
+                block_migration, disk_over_commit):
             raise exception.InvalidHypervisorType()
 
         self.stubs.Set(compute_api.API, 'update', fake_update)
-        self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
-                       'live_migration',
-                       fake_scheduler_api_live_migration)
+        self.stubs.Set(conductor_api.ComputeTaskAPI,
+                       'migrate_server',
+                       fake_migrate_server)
 
         res = req.get_response(app)
         self.assertEqual(res.status_int, 400)
@@ -270,16 +267,15 @@ class AdminActionsTest(test.TestCase):
                         task_state, expected_task_state):
             return None
 
-        def fake_scheduler_api_live_migration(context, dest,
-                                        block_migration=False,
-                                        disk_over_commit=False, instance=None,
-                                        instance_id=None, topic=None):
+        def fake_migrate_server(self, context, instance,
+                scheduler_hint, live, rebuild, flavor,
+                block_migration, disk_over_commit):
             raise exception.UnableToMigrateToSelf(self.UUID, host='host')
 
         self.stubs.Set(compute_api.API, 'update', fake_update)
-        self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
-                       'live_migration',
-                       fake_scheduler_api_live_migration)
+        self.stubs.Set(conductor_api.ComputeTaskAPI,
+                       'migrate_server',
+                       fake_migrate_server)
 
         res = req.get_response(app)
         self.assertEqual(res.status_int, 400)
@@ -308,16 +304,15 @@ class AdminActionsTest(test.TestCase):
                         task_state, expected_task_state):
             return None
 
-        def fake_scheduler_api_live_migration(context, dest,
-                                        block_migration=False,
-                                        disk_over_commit=False, instance=None,
-                                        instance_id=None, topic=None):
+        def fake_migrate_server(self, context, instance,
+                scheduler_hint, live, rebuild, flavor,
+                block_migration, disk_over_commit):
             raise exception.DestinationHypervisorTooOld()
 
         self.stubs.Set(compute_api.API, 'update', fake_update)
-        self.stubs.Set(scheduler_rpcapi.SchedulerAPI,
-                       'live_migration',
-                       fake_scheduler_api_live_migration)
+        self.stubs.Set(conductor_api.ComputeTaskAPI,
+                       'migrate_server',
+                       fake_migrate_server)
 
         res = req.get_response(app)
         self.assertEqual(res.status_int, 400)
diff --git a/nova/tests/conductor/test_conductor.py b/nova/tests/conductor/test_conductor.py
index c578e6de2..28d0074f2 100644
--- a/nova/tests/conductor/test_conductor.py
+++ b/nova/tests/conductor/test_conductor.py
@@ -1166,3 +1166,71 @@ class ConductorPolicyTest(test.TestCase):
 
         for key in keys:
             self.assertTrue(hasattr(instance, key))
+
+
+class _BaseTaskTestCase(object):
+    def setUp(self):
+        super(_BaseTaskTestCase, self).setUp()
+        self.user_id = 'fake'
+        self.project_id = 'fake'
+        self.context = FakeContext(self.user_id, self.project_id)
+
+    def test_migrate_server(self):
+        self.mox.StubOutWithMock(self.conductor_manager.scheduler_rpcapi,
+                                 'live_migration')
+        self.conductor_manager.scheduler_rpcapi.live_migration(self.context,
+            'block_migration', 'disk_over_commit', 'instance', 'destination')
+        self.mox.ReplayAll()
+        self.conductor.migrate_server(self.context, 'instance',
+            {'host': 'destination'}, True, False, None, 'block_migration',
+            'disk_over_commit')
+
+    def test_migrate_server_fails_with_non_live(self):
+        self.assertRaises(NotImplementedError, self.conductor.migrate_server,
+            self.context, None, None, False, False, None, None, None)
+
+    def test_migrate_server_fails_with_rebuild(self):
+        self.assertRaises(NotImplementedError, self.conductor.migrate_server,
+            self.context, None, None, True, True, None, None, None)
+
+    def test_migrate_server_fails_with_flavor(self):
+        self.assertRaises(NotImplementedError, self.conductor.migrate_server,
+            self.context, None, None, True, False, "dummy", None, None)
+
+
+class ConductorTaskTestCase(_BaseTaskTestCase, test.TestCase):
+    """ComputeTaskManager Tests."""
+    def setUp(self):
+        super(ConductorTaskTestCase, self).setUp()
+        self.conductor = conductor_manager.ComputeTaskManager()
+        self.conductor_manager = self.conductor
+
+
+class ConductorTaskRPCAPITestCase(_BaseTaskTestCase, test.TestCase):
+    """Conductor compute_task RPC namespace Tests."""
+    def setUp(self):
+        super(ConductorTaskRPCAPITestCase, self).setUp()
+        self.conductor_service = self.start_service(
+            'conductor', manager='nova.conductor.manager.ConductorManager')
+        self.conductor = conductor_rpcapi.ComputeTaskAPI()
+        service_manager = self.conductor_service.manager
+        self.conductor_manager = service_manager.compute_task_mgr
+
+
+class ConductorTaskAPITestCase(_BaseTaskTestCase, test.TestCase):
+    """Compute task API Tests."""
+    def setUp(self):
+        super(ConductorTaskAPITestCase, self).setUp()
+        self.conductor_service = self.start_service(
+            'conductor', manager='nova.conductor.manager.ConductorManager')
+        self.conductor = conductor_api.ComputeTaskAPI()
+        service_manager = self.conductor_service.manager
+        self.conductor_manager = service_manager.compute_task_mgr
+
+
+class ConductorLocalComputeTaskAPITestCase(ConductorTaskAPITestCase):
+    """Conductor LocalComputeTaskAPI Tests."""
+    def setUp(self):
+        super(ConductorLocalComputeTaskAPITestCase, self).setUp()
+        self.conductor = conductor_api.LocalComputeTaskAPI()
+        self.conductor_manager = self.conductor._manager._target