Add cache_image() support to the compute/{rpcapi,api,manager}

This provides the compute/api and compute/rpcapi plumbing to call
cache_image() on a given compute node. It also defines the status
information that we return to the caller to indicate what has been done.

Related to blueprint image-precache-support

Change-Id: If373fedb8d2e0dfc46b8ac5b018f8216aa5c643c

nova/compute/manager.py

@@ -521,7 +521,7 @@ class ComputeVirtAPI(virtapi.VirtAPI):
 class ComputeManager(manager.Manager):
     """Manages the running instances from creation to destruction."""
 
-    target = messaging.Target(version='5.3')
+    target = messaging.Target(version='5.4')
 
     def __init__(self, compute_driver=None, *args, **kwargs):
         """Load configuration options and connect to the hypervisor."""
@@ -9235,6 +9235,44 @@ class ComputeManager(manager.Manager):
 
         self.driver.manage_image_cache(context, filtered_instances)
 
+    def cache_images(self, context, image_ids):
+        """Ask the virt driver to pre-cache a set of base images.
+
+        :param context: The RequestContext
+        :param image_ids: The image IDs to be cached
+        :return: A dict, keyed by image-id where the values are one of:
+                 'cached' if the image was downloaded,
+                 'existing' if the image was already in the cache,
+                 'unsupported' if the virt driver does not support caching,
+                 'error' if the virt driver raised an exception.
+        """
+
+        results = {}
+
+        LOG.info('Caching %i image(s) by request', len(image_ids))
+        for image_id in image_ids:
+            try:
+                cached = self.driver.cache_image(context, image_id)
+                if cached:
+                    results[image_id] = 'cached'
+                else:
+                    results[image_id] = 'existing'
+            except NotImplementedError:
+                LOG.warning('Virt driver does not support image pre-caching;'
+                            ' ignoring request')
+                # NOTE(danms): Yes, technically we could short-circuit here to
+                # avoid trying the rest of the images, but it's very cheap to
+                # just keep hitting the NotImplementedError to keep the logic
+                # clean.
+                results[image_id] = 'unsupported'
+            except Exception as e:
+                results[image_id] = 'error'
+                LOG.error('Failed to cache image %(image_id)s: %(err)s',
+                          {'image_id': image_id,
+                           'err': e})
+
+        return results
+
     @periodic_task.periodic_task(spacing=CONF.instance_delete_interval)
     def _run_pending_deletes(self, context):
         """Retry any pending instance file deletes."""

nova/compute/rpcapi.py

@@ -370,6 +370,7 @@ class ComputeAPI(object):
         * 5.3 - Add migration and limits parameters to
                 check_can_live_migrate_destination(), and a new
                 drop_move_claim_at_destination() method
+        * 5.4 - Add cache_images() support
     '''
 
     VERSION_ALIASES = {
@@ -1234,3 +1235,17 @@ class ComputeAPI(object):
         cctxt = client.prepare(server=_compute_host(None, instance),
                 version=version)
         return cctxt.cast(ctxt, "trigger_crash_dump", instance=instance)
+
+    def cache_images(self, ctxt, host, image_ids):
+        version = '5.4'
+        client = self.router.client(ctxt)
+        if not client.can_send_version(version):
+            raise exception.NovaException('Compute RPC version pin does not '
+                                          'allow cache_images() to be called')
+        # This is a potentially very long-running call, so we provide the
+        # two timeout values which enables the call monitor in oslo.messaging
+        # so that this can run for extended periods.
+        cctxt = client.prepare(server=host, version=version,
+                               call_monitor_timeout=CONF.rpc_response_timeout,
+                               timeout=CONF.long_rpc_timeout)
+        return cctxt.call(ctxt, 'cache_images', image_ids=image_ids)

nova/conf/rpc.py

@@ -29,6 +29,7 @@ Operations with RPC calls that utilize this value:
 * live migration
 * scheduling
 * enabling/disabling a compute service
+* image pre-caching
 
 Related options:
 

nova/objects/service.py

@@ -31,7 +31,7 @@ LOG = logging.getLogger(__name__)
 
 
 # NOTE(danms): This is the global service version counter
-SERVICE_VERSION = 40
+SERVICE_VERSION = 41
 
 
 # NOTE(danms): This is our SERVICE_VERSION history. The idea is that any
@@ -161,6 +161,8 @@ SERVICE_VERSION_HISTORY = (
     # drop_move_claim_at_destination() method, and numa_live_migration
     # parameter to check_can_live_migrate_source()
     {'compute_rpc': '5.3'},
+    # Version 41: Add cache_images() to compute rpcapi (version 5.4)
+    {'compute_rpc': '5.4'},
 )
 
 

nova/tests/unit/compute/test_compute_mgr.py

@@ -5826,6 +5826,39 @@ class ComputeManagerUnitTestCase(test.NoDBTestCase,
                                details={'ovs_hybrid_plug': True})],
             [])
 
+    @mock.patch('nova.compute.manager.LOG')
+    def test_cache_images_unsupported(self, mock_log):
+        r = self.compute.cache_images(self.context, ['an-image'])
+        self.assertEqual({'an-image': 'unsupported'}, r)
+        mock_log.warning.assert_called_once_with(
+            'Virt driver does not support image pre-caching; ignoring request')
+
+    def test_cache_image_existing(self):
+        with mock.patch.object(self.compute.driver, 'cache_image') as c:
+            c.return_value = False
+            r = self.compute.cache_images(self.context, ['an-image'])
+            self.assertEqual({'an-image': 'existing'}, r)
+
+    def test_cache_image_downloaded(self):
+        with mock.patch.object(self.compute.driver, 'cache_image') as c:
+            c.return_value = True
+            r = self.compute.cache_images(self.context, ['an-image'])
+            self.assertEqual({'an-image': 'cached'}, r)
+
+    def test_cache_image_failed(self):
+        with mock.patch.object(self.compute.driver, 'cache_image') as c:
+            c.side_effect = test.TestingException('foo')
+            r = self.compute.cache_images(self.context, ['an-image'])
+            self.assertEqual({'an-image': 'error'}, r)
+
+    def test_cache_images_multi(self):
+        with mock.patch.object(self.compute.driver, 'cache_image') as c:
+            c.side_effect = [True, False]
+            r = self.compute.cache_images(self.context, ['one-image',
+                                                         'two-image'])
+            self.assertEqual({'one-image': 'cached',
+                              'two-image': 'existing'}, r)
+
 
 class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
     def setUp(self):

nova/tests/unit/compute/test_rpcapi.py

@@ -683,6 +683,22 @@ class ComputeRpcAPITestCase(test.NoDBTestCase):
                 request_spec=self.fake_request_spec_obj,
                 version='5.2')
 
+    def test_cache_image(self):
+        self._test_compute_api('cache_images', 'call',
+                               host='host', image_ids=['image'],
+                               call_monitor_timeout=60, timeout=1800,
+                               version='5.4')
+
+    def test_cache_image_pinned(self):
+        ctxt = context.RequestContext('fake_user', 'fake_project')
+        rpcapi = compute_rpcapi.ComputeAPI()
+        rpcapi.router.client = mock.Mock()
+        mock_client = mock.MagicMock()
+        rpcapi.router.client.return_value = mock_client
+        mock_client.can_send_version.return_value = False
+        self.assertRaises(exception.NovaException,
+                          rpcapi.cache_images, ctxt, 'host', ['image'])
+
     def test_unshelve_instance_old_compute(self):
         ctxt = context.RequestContext('fake_user', 'fake_project')
         rpcapi = compute_rpcapi.ComputeAPI()