RBD: Trash image when snapshots prevent deletion

RBD snapshots automatically end up in the trash
when deleted when using RBD image format v2.

However, they prevent images from being deleted.

Instead, move RBD images to the trash in this
situation.  This trash must be purged by a scheduled
rbd trash purge operation outside of Glance.

This passes new cinder<->image dependency tests in
I5fee2395195 when using ceph require-min-compat-client
of "mimic", which enables clone v2.

Change-Id: I34dcd90a09d43127ff2e8b477750c70f3cc01113

glance_store/_drivers/rbd.py

@@ -471,14 +471,6 @@ class Store(driver.Store):
                     if snapshot_name is not None:
                         with rbd.Image(ioctx, image_name) as image:
                             try:
-                                # NOTE(abhishekk): Check whether snapshot
-                                # has any external references
-                                if self._snapshot_has_external_reference(
-                                        image, snapshot_name):
-                                    raise rbd.ImageBusy(
-                                        "Image snapshot has external "
-                                        "references.")
-
                                 self._unprotect_snapshot(image, snapshot_name)
                                 image.remove_snap(snapshot_name)
                             except rbd.ImageNotFound as exc:
@@ -502,6 +494,13 @@ class Store(driver.Store):
                                    "It has snapshot(s) left.") %
                                {'img_name': image_name})
                     LOG.warning(log_msg)
+                    with rbd.Image(ioctx, image_name) as image:
+                        try:
+                            rbd.RBD().trash_move(ioctx, image_name)
+                            LOG.debug('Moved %s to trash', image_name)
+                        except rbd.ImageBusy:
+                            raise exceptions.InUseByStore()
+                        return
                     raise exceptions.HasSnapshot()
                 except rbd.ImageBusy:
                     log_msg = (_LW("Remove image %(img_name)s failed. "

glance_store/tests/unit/test_multistore_rbd.py

@@ -172,6 +172,9 @@ class MockRBD(object):
         def clone(self, *args, **kwargs):
             raise NotImplementedError()
 
+        def trash_move(self, *args, **kwargs):
+            pass
+
     RBD_FEATURE_LAYERING = 1
 
 
@@ -427,10 +430,9 @@ class TestMultiStore(base.MultiStoreBaseTest,
         with mock.patch.object(MockRBD.Image, 'list_children') as mocked:
             mocked.return_value = True
 
-            self.assertRaises(exceptions.InUseByStore,
-                              self.store._delete_image,
-                              'fake_pool', self.location.image,
-                              snapshot_name='snap')
+            self.store._delete_image('fake_pool',
+                                     self.location.image,
+                                     snapshot_name='snap')
 
     def test_delete_image_w_snap_exc_image_has_snap(self):
         def _fake_remove(*args, **kwargs):
@@ -439,8 +441,8 @@ class TestMultiStore(base.MultiStoreBaseTest,
 
         with mock.patch.object(MockRBD.RBD, 'remove') as remove:
             remove.side_effect = _fake_remove
-            self.assertRaises(exceptions.HasSnapshot, self.store._delete_image,
-                              'fake_pool', self.location.image)
+            self.store._delete_image('fake_pool',
+                                     self.location.image)
 
             self.called_commands_expected = ['remove']
 

glance_store/tests/unit/test_rbd_store.py

@@ -173,6 +173,9 @@ class MockRBD(object):
         def clone(self, *args, **kwargs):
             raise NotImplementedError()
 
+        def trash_move(self, *args, **kwargs):
+            pass
+
     RBD_FEATURE_LAYERING = 1
 
 
@@ -633,23 +636,45 @@ class TestStore(base.StoreBaseTest,
         with mock.patch.object(MockRBD.Image, 'list_children') as mocked:
             mocked.return_value = True
 
-            self.assertRaises(exceptions.InUseByStore,
-                              self.store._delete_image,
-                              'fake_pool', self.location.image,
-                              snapshot_name='snap')
+            self.store._delete_image('fake_pool',
+                                     self.location.image,
+                                     snapshot_name='snap')
 
     def test_delete_image_w_snap_exc_image_has_snap(self):
         def _fake_remove(*args, **kwargs):
             self.called_commands_actual.append('remove')
             raise MockRBD.ImageHasSnapshots()
 
+        mock.patch.object(MockRBD.RBD, 'trash_move').start()
+
         with mock.patch.object(MockRBD.RBD, 'remove') as remove:
             remove.side_effect = _fake_remove
-            self.assertRaises(exceptions.HasSnapshot, self.store._delete_image,
-                              'fake_pool', self.location.image)
+            self.store._delete_image('fake_pool',
+                                     self.location.image)
 
             self.called_commands_expected = ['remove']
 
+        MockRBD.RBD.trash_move.assert_called_once_with(mock.ANY, 'fake_image')
+
+    def test_delete_image_w_snap_exc_image_has_snap_2(self):
+        def _fake_remove(*args, **kwargs):
+            self.called_commands_actual.append('remove')
+            raise MockRBD.ImageHasSnapshots()
+
+        mock.patch.object(MockRBD.RBD, 'trash_move',
+                          side_effect=MockRBD.ImageBusy).start()
+
+        with mock.patch.object(MockRBD.RBD, 'remove') as remove:
+            remove.side_effect = _fake_remove
+            self.assertRaises(exceptions.InUseByStore,
+                              self.store._delete_image,
+                              'fake_pool',
+                              self.location.image)
+
+            self.called_commands_expected = ['remove']
+
+        MockRBD.RBD.trash_move.assert_called_once_with(mock.ANY, 'fake_image')
+
     def test_get_partial_image(self):
         loc = g_location.Location('test_rbd_store', rbd_store.StoreLocation,
                                   self.conf, store_specs=self.store_specs)