Catch ImageBusy exception when deleting rbd volume

If we try to delete an rbd volume that has 'watchers' on it
i.e. client connections that have not yet been closed
possibly because a client crashed, the remove() will throw an
ImageBusy exception. We now catch this exception and raise
VolumeIsBusy with a useful message.

If the volume delete fails in this way it will now stay as
'available' instead of going to 'error_deleting' so that the
delete can be retried (since it is expected to work on a
retry after waiting for the connection to timeout).

Change-Id: I5bc9a5f71bdb0f9c5d12b5577e68377e66561f5b
Closes-bug: 1256259

cinder/tests/test_rbd.py

@@ -17,6 +17,7 @@
 
 
 import contextlib
+import mock
 import mox
 import os
 import tempfile
@@ -104,6 +105,9 @@ def fake_execute(*args, **kwargs):
                                        rbd=self.rbd)
         self.driver.set_initialized()
 
+    def tearDown(self):
+        super(RBDTestCase, self).tearDown()
+
     def test_create_volume(self):
         name = u'volume-00000001'
         size = 1
@@ -125,36 +129,77 @@ def test_create_volume(self):
 
         self.driver.create_volume(volume)
 
-    def test_delete_volume(self):
+    @mock.patch('cinder.volume.drivers.rbd.rados')
+    @mock.patch('cinder.volume.drivers.rbd.rbd')
+    def test_delete_volume(self, _mock_rbd, _mock_rados):
         name = u'volume-00000001'
         volume = dict(name=name)
 
-        # Setup librbd stubs
-        self.stubs.Set(self.driver, 'rados', mock_rados)
-        self.stubs.Set(self.driver, 'rbd', mock_rbd)
-
-        class mock_client(object):
-            def __init__(self, *args, **kwargs):
-                self.ioctx = None
-
-            def __enter__(self, *args, **kwargs):
-                return self
-
-            def __exit__(self, type_, value, traceback):
-                pass
+        _mock_rbd.Image = mock_rbd.Image
+        _mock_rbd.Image.list_snaps = mock.Mock()
+        _mock_rbd.Image.list_snaps.return_value = []
+        _mock_rbd.Image.unprotect_snap = mock.Mock()
+
+        _mock_rbd.RBD = mock_rbd.RBD
+        _mock_rbd.RBD.remove = mock.Mock()
+
+        self.driver.rbd = _mock_rbd
+        self.driver.rados = _mock_rados
+
+        mpo = mock.patch.object
+        with mpo(driver, 'RADOSClient') as mock_rados_client:
+            with mpo(self.driver, '_get_clone_info') as mock_get_clone_info:
+                mock_get_clone_info.return_value = (None, None, None)
+                with mpo(self.driver,
+                         '_delete_backup_snaps') as mock_del_backup_snaps:
+                    self.driver.delete_volume(volume)
+
+                    self.assertTrue(mock_get_clone_info.called)
+                    self.assertTrue(_mock_rbd.Image.list_snaps.called)
+                    self.assertTrue(mock_rados_client.called)
+                    self.assertTrue(mock_del_backup_snaps.called)
+                    self.assertFalse(mock_rbd.Image.unprotect_snap.called)
+                    self.assertTrue(_mock_rbd.RBD.remove.called)
+
+    @mock.patch('cinder.volume.drivers.rbd.rados')
+    @mock.patch('cinder.volume.drivers.rbd.rbd')
+    def test_delete_busy_volume(self, _mock_rbd, _mock_rados):
+        name = u'volume-00000001'
+        volume = dict(name=name)
 
-        self.stubs.Set(driver, 'RADOSClient', mock_client)
+        _mock_rbd.Image = mock_rbd.Image
+        _mock_rbd.Image.list_snaps = mock.Mock()
+        _mock_rbd.Image.list_snaps.return_value = []
+        _mock_rbd.Image.unprotect_snap = mock.Mock()
 
-        self.stubs.Set(self.driver, '_get_backup_snaps',
-                       lambda *args: None)
-        self.stubs.Set(self.driver.rbd.Image, 'list_snaps',
-                       lambda *args: [])
-        self.stubs.Set(self.driver.rbd.Image, 'parent_info',
-                       lambda *args: (None, None, None))
-        self.stubs.Set(self.driver.rbd.Image, 'unprotect_snap',
-                       lambda *args: None)
+        class MyMockException(Exception):
+            pass
 
-        self.driver.delete_volume(volume)
+        _mock_rbd.RBD = mock_rbd.RBD
+        _mock_rbd.ImageBusy = MyMockException
+        _mock_rbd.RBD.remove = mock.Mock()
+        _mock_rbd.RBD.remove.side_effect = _mock_rbd.ImageBusy
+
+        self.driver.rbd = _mock_rbd
+        self.driver.rados = _mock_rados
+
+        mpo = mock.patch.object
+        with mpo(driver, 'RADOSClient') as mock_rados_client:
+            with mpo(self.driver, '_get_clone_info') as mock_get_clone_info:
+                mock_get_clone_info.return_value = (None, None, None)
+                with mpo(self.driver,
+                         '_delete_backup_snaps') as mock_del_backup_snaps:
+
+                    self.assertRaises(exception.VolumeIsBusy,
+                                      self.driver.delete_volume,
+                                      volume)
+
+                    self.assertTrue(mock_get_clone_info.called)
+                    self.assertTrue(_mock_rbd.Image.list_snaps.called)
+                    self.assertTrue(mock_rados_client.called)
+                    self.assertTrue(mock_del_backup_snaps.called)
+                    self.assertFalse(mock_rbd.Image.unprotect_snap.called)
+                    self.assertTrue(_mock_rbd.RBD.remove.called)
 
     def test_create_snapshot(self):
         vol_name = u'volume-00000001'

cinder/volume/drivers/rbd.py

@@ -623,7 +623,18 @@ def delete_volume(self, volume):
 
             if clone_snap is None:
                 LOG.debug(_("deleting rbd volume %s") % (volume_name))
-                self.rbd.RBD().remove(client.ioctx, volume_name)
+                try:
+                    self.rbd.RBD().remove(client.ioctx, volume_name)
+                except self.rbd.ImageBusy:
+                    msg = (_("ImageBusy error raised while deleting rbd "
+                             "volume. This may have been caused by a "
+                             "connection from a client that has crashed and, "
+                             "if so, may be resolved by retrying the delete "
+                             "after 30 seconds has elapsed."))
+                    LOG.error(msg)
+                    # Now raise this so that volume stays available so that we
+                    # delete can be retried.
+                    raise exception.VolumeIsBusy(msg, volume_name=volume_name)
 
                 # If it is a clone, walk back up the parent chain deleting
                 # references.