Add shared_storage_test methods to compute rpcapi.

Part of bug 1006467.

This patch adds 3 methods used to check if shared storage works between
two compute hosts to the compute rpcapi.  These methods are used by the
scheduler.

This also cuts down on some duplicated code in test_scheduler.

Change-Id: If0bf7f7fdd2beb0e5280ae52daf049372f8bc9e2

4 files changed, 67 insertions, 95 deletions

diff --git a/nova/compute/rpcapi.py b/nova/compute/rpcapi.py
index 155c23abf..8c25906c9 100644
--- a/nova/compute/rpcapi.py
+++ b/nova/compute/rpcapi.py
@@ -84,6 +84,16 @@ class ComputeAPI(nova.rpc.proxy.RpcProxy):
                 mountpoint=mountpoint),
                 topic=self._compute_topic(ctxt, None, instance))
 
+    def check_shared_storage_test_file(self, ctxt, filename, host):
+        return self.call(ctxt, self.make_msg('check_shared_storage_test_file',
+                filename=filename),
+                topic=self._compute_topic(ctxt, host, None))
+
+    def cleanup_shared_storage_test_file(self, ctxt, filename, host):
+        self.cast(ctxt, self.make_msg('cleanup_shared_storage_test_file',
+                filename=filename),
+                topic=self._compute_topic(ctxt, host, None))
+
     def compare_cpu(self, ctxt, cpu_info, host):
         return self.call(ctxt, self.make_msg('compare_cpu', cpu_info=cpu_info),
                 topic=self._compute_topic(ctxt, host, None))
@@ -95,6 +105,11 @@ class ComputeAPI(nova.rpc.proxy.RpcProxy):
                 instance_uuid=instance['uuid'], migration_id=migration_id),
                 topic=self._compute_topic(ctxt, host, instance))
 
+    def create_shared_storage_test_file(self, ctxt, host):
+        return self.call(ctxt,
+                self.make_msg('create_shared_storage_test_file'),
+                topic=self._compute_topic(ctxt, host, None))
+
     def detach_volume(self, ctxt, instance, volume_id):
         self.cast(ctxt, self.make_msg('detach_volume',
                 instance_uuid=instance['uuid'], volume_id=volume_id),
diff --git a/nova/scheduler/driver.py b/nova/scheduler/driver.py
index 94c3395fe..bcdc6ce10 100644
--- a/nova/scheduler/driver.py
+++ b/nova/scheduler/driver.py
@@ -489,21 +489,17 @@ class Scheduler(object):
         """
 
         src = instance_ref['host']
-        dst_t = rpc.queue_get_for(context, FLAGS.compute_topic, dest)
-        src_t = rpc.queue_get_for(context, FLAGS.compute_topic, src)
 
-        filename = rpc.call(context, dst_t,
-                            {"method": 'create_shared_storage_test_file'})
+        filename = self.compute_rpcapi.create_shared_storage_test_file(context,
+                dest)
 
         try:
             # make sure existence at src host.
-            ret = rpc.call(context, src_t,
-                        {"method": 'check_shared_storage_test_file',
-                        "args": {'filename': filename}})
+            ret = self.compute_rpcapi.check_shared_storage_test_file(context,
+                    filename, src)
 
         finally:
-            rpc.cast(context, dst_t,
-                    {"method": 'cleanup_shared_storage_test_file',
-                    "args": {'filename': filename}})
+            self.compute_rpcapi.cleanup_shared_storage_test_file(context,
+                    filename, dest)
 
         return ret
diff --git a/nova/tests/compute/test_rpcapi.py b/nova/tests/compute/test_rpcapi.py
index dabb1f4b2..47fb10645 100644
--- a/nova/tests/compute/test_rpcapi.py
+++ b/nova/tests/compute/test_rpcapi.py
@@ -106,6 +106,14 @@ class ComputeRpcAPITestCase(test.TestCase):
         self._test_compute_api('attach_volume', 'cast',
                 instance=self.fake_instance, volume_id='id', mountpoint='mp')
 
+    def test_check_shared_storage_test_file(self):
+        self._test_compute_api('check_shared_storage_test_file', 'call',
+                filename='fn', host='host')
+
+    def test_cleanup_shared_storage_test_file(self):
+        self._test_compute_api('cleanup_shared_storage_test_file', 'cast',
+                filename='fn', host='host')
+
     def test_compare_cpu(self):
         self._test_compute_api('compare_cpu', 'call', cpu_info='info',
                 host='host')
@@ -118,6 +126,10 @@ class ComputeRpcAPITestCase(test.TestCase):
         self._test_compute_api('confirm_resize', 'call',
                 instance=self.fake_instance, migration_id='id', host='host')
 
+    def test_create_shared_storage_test_file(self):
+        self._test_compute_api('create_shared_storage_test_file', 'call',
+                host='host')
+
     def test_detach_volume(self):
         self._test_compute_api('detach_volume', 'cast',
                 instance=self.fake_instance, volume_id='id')
diff --git a/nova/tests/scheduler/test_scheduler.py b/nova/tests/scheduler/test_scheduler.py
index 2ad79eed6..c8f02d357 100644
--- a/nova/tests/scheduler/test_scheduler.py
+++ b/nova/tests/scheduler/test_scheduler.py
@@ -469,6 +469,29 @@ class SchedulerTestCase(test.TestCase):
                 block_migration=block_migration,
                 disk_over_commit=disk_over_commit)
 
+    def _check_shared_storage(self, dest, instance, check_result):
+        tmp_filename = 'test-filename'
+        rpc.queue_get_for(self.context, FLAGS.compute_topic,
+                dest).AndReturn('dest_queue')
+        rpc.call(self.context, 'dest_queue',
+                {'method': 'create_shared_storage_test_file',
+                 'args': {},
+                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
+                ).AndReturn(tmp_filename)
+        rpc.queue_get_for(self.context, FLAGS.compute_topic,
+                instance['host']).AndReturn('src_queue')
+        rpc.call(self.context, 'src_queue',
+                {'method': 'check_shared_storage_test_file',
+                 'args': {'filename': tmp_filename},
+                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION}, None
+                ).AndReturn(check_result)
+        rpc.queue_get_for(self.context, FLAGS.compute_topic,
+                dest).AndReturn('dest_queue')
+        rpc.cast(self.context, 'dest_queue',
+                {'method': 'cleanup_shared_storage_test_file',
+                 'args': {'filename': tmp_filename},
+                 'version': compute_rpcapi.ComputeAPI.RPC_API_VERSION})
+
     def test_live_migration_all_checks_pass(self):
         """Test live migration when all checks pass."""
 
@@ -514,21 +537,9 @@ class SchedulerTestCase(test.TestCase):
                  None).AndReturn(
                         json.dumps([{'disk_size': 1024 * (1024 ** 3)}]))
 
-        # Common checks (shared storage ok, same hypervisor,e tc)
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(False)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        # Common checks (shared storage ok, same hypervisor, etc)
+        self._check_shared_storage(dest, instance, False)
+
         db.service_get_all_compute_by_host(self.context, dest).AndReturn(
                 [{'compute_node': [{'hypervisor_type': 'xen',
                                     'hypervisor_version': 1}]}])
@@ -758,20 +769,7 @@ class SchedulerTestCase(test.TestCase):
         self.driver._live_migration_dest_check(self.context, instance,
                 dest, block_migration, disk_over_commit)
 
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(False)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        self._check_shared_storage(dest, instance, False)
 
         self.mox.ReplayAll()
         self.assertRaises(exception.InvalidSharedStorage,
@@ -800,20 +798,7 @@ class SchedulerTestCase(test.TestCase):
         self.driver._live_migration_dest_check(self.context, instance,
                 dest, block_migration, disk_over_commit)
 
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(False)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        self._check_shared_storage(dest, instance, False)
 
         self.mox.ReplayAll()
         self.assertRaises(exception.InvalidSharedStorage,
@@ -841,20 +826,8 @@ class SchedulerTestCase(test.TestCase):
         self.driver._live_migration_dest_check(self.context, instance,
                 dest, block_migration, disk_over_commit)
 
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(True)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        self._check_shared_storage(dest, instance, True)
+
         db.service_get_all_compute_by_host(self.context, dest).AndReturn(
                 [{'compute_node': [{'hypervisor_type': 'xen',
                                     'hypervisor_version': 1}]}])
@@ -890,20 +863,8 @@ class SchedulerTestCase(test.TestCase):
         self.driver._live_migration_dest_check(self.context, instance,
                 dest, block_migration, disk_over_commit)
 
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(True)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        self._check_shared_storage(dest, instance, True)
+
         db.service_get_all_compute_by_host(self.context, dest).AndReturn(
                 [{'compute_node': [{'hypervisor_type': 'xen',
                                     'hypervisor_version': 1}]}])
@@ -938,20 +899,8 @@ class SchedulerTestCase(test.TestCase):
         self.driver._live_migration_dest_check(self.context, instance,
                 dest, block_migration, disk_over_commit)
 
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                dest).AndReturn('dest_queue')
-        rpc.queue_get_for(self.context, FLAGS.compute_topic,
-                instance['host']).AndReturn('src_queue')
-        tmp_filename = 'test-filename'
-        rpc.call(self.context, 'dest_queue',
-                {'method': 'create_shared_storage_test_file'}
-                ).AndReturn(tmp_filename)
-        rpc.call(self.context, 'src_queue',
-                {'method': 'check_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}}).AndReturn(True)
-        rpc.cast(self.context, 'dest_queue',
-                {'method': 'cleanup_shared_storage_test_file',
-                 'args': {'filename': tmp_filename}})
+        self._check_shared_storage(dest, instance, True)
+
         db.service_get_all_compute_by_host(self.context, dest).AndReturn(
                 [{'compute_node': [{'hypervisor_type': 'xen',
                                     'hypervisor_version': 1}]}])