Fix: Create new cache entry when xtremio reaches snap limit

When we reach xtremio_volumes_per_glance_cache, it is not able
to create more clones for the image-volume therefore requires to
create a new cache entry.
Keeping in mind the case in [1], we can get CinderException for
various reasons from different drivers during the clone operation.
So we define a new exception for the xtremio driver to force create
a new cache entry when the limit reaches and not enforce the same
for other drivers.

[1] https://bugs.launchpad.net/cinder/+bug/1552734

Closes-Bug: #1858169

Change-Id: I2bf964d5a7b2048db9be1ea3eb97cd517e112c5b
(cherry picked from commit d22e54c254)

cinder/exception.py

@@ -1117,3 +1117,8 @@ class CinderAcceleratorError(CinderException):
                 "Command %(cmd)s execution failed.\n"
                 "%(description)s\n"
                 "Reason: %(reason)s")
+
+
+class SnapshotLimitReached(CinderException):
+    message = _("Exceeded the configured limit of "
+                "%(set_limit)s snapshots per volume.")

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

@@ -1559,53 +1559,52 @@ class CreateVolumeFlowManagerImageCacheTestCase(test.TestCase):
         image_location = 'someImageLocationStr'
         image_id = fakes.IMAGE_ID
         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_driver.clone_image.return_value = (None, False)
-        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,
             self.mock_db,
             self.mock_driver,
             image_volume_cache=self.mock_cache
         )
-
+        if cloning_supported:
-        model_update = manager._create_from_image_cache_or_download(
+            mock_create_from_src.side_effect = exception.SnapshotLimitReached(
-            self.ctxt,
+                'Error during cloning')
-            volume,
+            self.assertRaises(
-            image_location,
+                exception.SnapshotLimitReached,
-            image_id,
+                manager._create_from_image,
-            image_meta,
+                self.ctxt,
-            self.mock_image_service,
+                volume,
-            update_cache=False)
+                image_location,
+                image_id,
+                image_meta,
+                self.mock_image_service)
+        else:
+            mock_create_from_src.side_effect = NotImplementedError(
+                'Driver does not support clone')
+            model_update = manager._create_from_image(
+                self.ctxt,
+                volume,
+                image_location,
+                image_id,
+                image_meta,
+                self.mock_image_service)
+            mock_create_from_img_dl.assert_called_once()
+            self.assertEqual(mock_create_from_img_dl.return_value,
+                             model_update)
 
         # Ensure cloning was attempted and that it failed
         mock_create_from_src.assert_called_once()
-        mock_create_from_img_dl.assert_called_once()
+        with mock.patch(
-        self.assertEqual(mock_create_from_img_dl.return_value, model_update)
+                'cinder.volume.flows.manager.create_volume.'
-
+                'CreateVolumeFromSpecTask') as volume_manager:
-        # Ensure a new cache entry is created when cloning fails, but
+            (volume_manager.CreateVolumeFromSpecTask.
-        # only when the driver supports cloning.
+             _create_from_image_cache_or_download.called_once())
-        if cloning_supported:
+            (volume_manager.CreateVolumeFromSpecTask.
-            (self.mock_volume_manager.
+             _create_from_image_cache.called_once())
-             _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/drivers/dell_emc/xtremio.py

@@ -552,8 +552,7 @@ class XtremIOVolumeDriver(san.SanDriver):
                                                             src_vref['id'])
         limit = self.configuration.safe_get('xtremio_volumes_per_glance_cache')
         if cache and limit and limit > 0 and limit <= vol['num-of-dest-snaps']:
-            raise exception.CinderException('Exceeded the configured limit of '
+            raise exception.SnapshotLimitReached(set_limit=limit)
-                                            '%d snapshots per volume' % limit)
         try:
             self.client.create_snapshot(src_vref['id'], volume['id'])
         except XtremIOSnapshotsLimitExceeded as e:

cinder/volume/flows/manager/create_volume.py

@@ -772,6 +772,14 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
                     cache_entry['volume_id']
                 )
                 return model_update, True
+        except exception.SnapshotLimitReached:
+            # If this exception occurred when cloning the image-volume,
+            # it is because the image-volume reached its snapshot limit.
+            # Delete current cache entry and create a "fresh" entry
+            # NOTE: This will not delete the existing image-volume and
+            # only delete the cache entry
+            with excutils.save_and_reraise_exception():
+                self.image_volume_cache.evict(context, cache_entry)
         except NotImplementedError:
             LOG.warning('Backend does not support creating image-volume '
                         'clone. Image will be downloaded from Glance.')
@@ -855,17 +863,18 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
                         image_id,
                         image_meta
                     )
+                except exception.SnapshotLimitReached:
+                    # This exception will be handled by the caller's
+                    # (_create_from_image) retry decorator
+                    with excutils.save_and_reraise_exception():
+                        LOG.debug("Snapshot limit reached. Creating new "
+                                  "image-volume.")
                 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
@@ -957,6 +966,7 @@ class CreateVolumeFromSpecTask(flow_utils.CinderTask):
 
         return model_update
 
+    @utils.retry(exception.SnapshotLimitReached, retries=1)
     def _create_from_image(self, context, volume,
                            image_location, image_id, image_meta,
                            image_service, **kwargs):