Allow deletion of backups where the service is None

If a backup is created while both cinder-backup and rabbitmq
are not running, backup records are created in the database with a
service entry of None.

There is no actual backup data created on the service, since the service
isn't running. This fix allows removal of these records without an error
when a delete backup request is received.

This fix also ensures backup status is set to error in the event of
an exception during a delete.

Fixes bug #1162908

Change-Id: I5771747a00a70621f7cc101f8c1da2f613e83cdf

diff --git a/cinder/backup/manager.py b/cinder/backup/manager.py
index 11843cd5b28babd65dcfe0f7754d80eb6b40ece4..1af3576795e6412a8d3f649952c522ff0f3d583a 100755 (executable)
--- a/cinder/backup/manager.py
+++ b/cinder/backup/manager.py
@@ -238,23 +238,26 @@ class BackupManager(manager.SchedulerDependentManager):
             raise exception.InvalidBackup(reason=err)
 
         backup_service = backup['service']
-        configured_service = FLAGS.backup_service
-        if backup_service != configured_service:
-            err = _('delete_backup aborted, the backup service currently'
-                    ' configured [%(configured_service)s] is not the'
-                    ' backup service that was used to create this'
-                    ' backup [%(backup_service)s]') % locals()
-            self.db.backup_update(context, backup_id, {'status': 'available'})
-            raise exception.InvalidBackup(reason=err)
-
-        try:
-            backup_service = self.service.get_backup_service(context)
-            backup_service.delete(backup)
-        except Exception as err:
-            with excutils.save_and_reraise_exception():
-                self.db.backup_update(context, backup_id, {'status': 'error',
-                                                           'fail_reason':
-                                                           unicode(err)})
+        if backup_service is not None:
+            configured_service = FLAGS.backup_service
+            if backup_service != configured_service:
+                err = _('delete_backup aborted, the backup service currently'
+                        ' configured [%(configured_service)s] is not the'
+                        ' backup service that was used to create this'
+                        ' backup [%(backup_service)s]') % locals()
+                self.db.backup_update(context, backup_id,
+                                      {'status': 'error'})
+                raise exception.InvalidBackup(reason=err)
+
+            try:
+                backup_service = self.service.get_backup_service(context)
+                backup_service.delete(backup)
+            except Exception as err:
+                with excutils.save_and_reraise_exception():
+                    self.db.backup_update(context, backup_id,
+                                          {'status': 'error',
+                                           'fail_reason':
+                                           unicode(err)})
 
         context = context.elevated()
         self.db.backup_destroy(context, backup_id)

diff --git a/cinder/tests/test_backup.py b/cinder/tests/test_backup.py
index b97a76f4c12586e9eac63899ff552af4663c242a..8ffe9006c0a17f29f65a894b26ba2cd22fda429e 100644 (file)
--- a/cinder/tests/test_backup.py
+++ b/cinder/tests/test_backup.py
@@ -318,7 +318,16 @@ class BackupTestCase(test.TestCase):
                           self.ctxt,
                           backup_id)
         backup = db.backup_get(self.ctxt, backup_id)
-        self.assertEquals(backup['status'], 'available')
+        self.assertEquals(backup['status'], 'error')
+
+    def test_delete_backup_with_no_service(self):
+        """Test error handling when attempting a delete of a backup
+        with no service defined for that backup, relates to bug #1162908"""
+        vol_id = self._create_volume_db_entry(size=1)
+        backup_id = self._create_backup_db_entry(status='deleting',
+                                                 volume_id=vol_id)
+        db.backup_update(self.ctxt, backup_id, {'service': None})
+        self.backup_mgr.delete_backup(self.ctxt, backup_id)
 
     def test_delete_backup(self):
         """Test normal backup deletion"""