Create new image volume cache entry when cloning fails

This patch fixes the image volume cache so that a new cache entry is
created whenever cloning an existing entry fails (which can happen, for
example, when a cached volume reaches its snapshot limit). This restores
the original behavior, which was broken by [1].

[1] I547fb4bcdd4783225b8ca96d157c61ca3bcf4ef4

Closes-Bug: #1801595
Change-Id: Ib5947e2c7300730adb851ad58e898a29f2b88525

cinder/tests/unit/volume/flows/test_create_volume_flow.py

@@ -1548,19 +1548,34 @@ class CreateVolumeFlowManagerImageCacheTestCase(test.TestCase):
             image_meta=image_meta
         )
 
-    @ddt.data(
-        NotImplementedError('Driver does not support clone'),
-        exception.CinderException('Error during cloning'))
+    @ddt.data(False, True)
+    @mock.patch('cinder.image.image_utils.check_available_space')
+    @mock.patch('cinder.image.image_utils.qemu_img_info')
     def test_create_from_image_clone_failure(
-            self, effect, mock_get_internal_context,
+            self, cloning_supported, mock_qemu_info, mock_check_space,
+            mock_get_internal_context,
             mock_create_from_img_dl, mock_create_from_src,
             mock_handle_bootable, mock_fetch_img):
-        mock_get_internal_context.return_value = None
-        volume = fake_volume.fake_volume_obj(self.ctxt)
-        mock_create_from_src.side_effect = effect
-
+        image_location = 'someImageLocationStr'
         image_id = fakes.IMAGE_ID
-        image_meta = {'virtual_size': '1073741824'}
+        image_meta = mock.MagicMock()
+        image_info = imageutils.QemuImgInfo()
+        image_info.virtual_size = '1073741824'
+        mock_qemu_info.return_value = image_info
+
+        volume = fake_volume.fake_volume_obj(self.ctxt, size=1,
+                                             host='foo@bar#pool')
+        image_volume = fake_volume.fake_db_volume(size=2)
+        self.mock_db.volume_create.return_value = image_volume
+
+        self.flags(verify_glance_signatures='disabled')
+
+        if cloning_supported:
+            mock_create_from_src.side_effect = exception.CinderException(
+                'Error during cloning')
+        else:
+            mock_create_from_src.side_effect = NotImplementedError(
+                'Driver does not support clone')
 
         manager = create_volume_manager.CreateVolumeFromSpecTask(
             self.mock_volume_manager,
@@ -1569,14 +1584,28 @@ class CreateVolumeFlowManagerImageCacheTestCase(test.TestCase):
             image_volume_cache=self.mock_cache
         )
 
-        model, result = manager._create_from_image_cache(self.ctxt,
-                                                         None,
-                                                         volume,
-                                                         image_id,
-                                                         image_meta)
+        model_update = manager._create_from_image_cache_or_download(
+            self.ctxt,
+            volume,
+            image_location,
+            image_id,
+            image_meta,
+            self.mock_image_service,
+            update_cache=False)
 
-        self.assertIsNone(model)
-        self.assertFalse(result)
+        # Ensure cloning was attempted and that it failed
+        mock_create_from_src.assert_called_once()
+        mock_create_from_img_dl.assert_called_once()
+        self.assertEqual(mock_create_from_img_dl.return_value, model_update)
+
+        # Ensure a new cache entry is created when cloning fails, but
+        # only when the driver supports cloning.
+        if cloning_supported:
+            (self.mock_volume_manager.
+             _create_image_cache_volume_entry.assert_called_once())
+        else:
+            (self.mock_volume_manager.
+             _create_image_cache_volume_entry.assert_not_called())
 
     @mock.patch('cinder.volume.flows.manager.create_volume.'
                 'CreateVolumeFromSpecTask.'

cinder/volume/flows/manager/create_volume.py

@@ -622,10 +622,6 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
         except NotImplementedError:
             LOG.warning('Backend does not support creating image-volume '
                         'clone. Image will be downloaded from Glance.')
-        except exception.CinderException as e:
-            LOG.warning('Failed to create volume from image-volume cache, '
-                        'image will be downloaded from Glance. Error: '
-                        '%(exception)s', {'exception': e})
         return None, False
 
     @coordination.synchronized('{image_id}')
@@ -698,13 +694,25 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
                 LOG.info('Unable to get Cinder internal context, will '
                          'not use image-volume cache.')
             else:
-                model_update, cloned = self._create_from_image_cache(
-                    context,
-                    internal_context,
-                    volume,
-                    image_id,
-                    image_meta
-                )
+                try:
+                    model_update, cloned = self._create_from_image_cache(
+                        context,
+                        internal_context,
+                        volume,
+                        image_id,
+                        image_meta
+                    )
+                except exception.CinderException as e:
+                    LOG.warning('Failed to create volume from image-volume '
+                                'cache, image will be downloaded from Glance. '
+                                'Error: %(exception)s',
+                                {'exception': e})
+
+                    # If an exception occurred when cloning the image-volume,
+                    # it may be the image-volume reached its snapshot limit.
+                    # Create another "fresh" cache entry.
+                    update_cache = True
+
                 # Don't cache unless directed.
                 if not cloned and update_cache:
                     should_create_cache_entry = True