Check cinder-backup service before "backing-up"

If cinder-backup service is not enabled, "cinder backup-create"
command fails like the following. As the result, the volume status
has been changed to "backing-up" in spite of not backing-up.

$ cinder backup-create f48aa6ae-4c35-4a6a-a393-5a5cf610945a
ERROR: Service cinder-backup could not be found.
$ cinder list
+--+------------+--------------+------+-------------+----------+-------------+
|ID|   Status   | Display Name | Size | Volume Type | Bootable | Attached to |
+--+------------+--------------+------+-------------+----------+-------------+
|..| backing-up |   vol-test   |  1   |     None    |  False   |             |
+--+------------+--------------+------+-------------+----------+-------------+
$

To avoid this situation, this patch moves the cinder-backup check
before changing a volume status to "backing-up".

Fixes bug #1221012

Change-Id: I42ad41e1cfb6fdb7feebe39a9a5f0356a41d7838

cinder/backup/api.py

@@ -84,7 +84,7 @@ class API(base.Base):
 
         return backups
 
-    def _check_backup_service(self, volume):
+    def _is_backup_service_enabled(self, volume):
         """Check if there is an backup service available"""
         topic = CONF.backup_topic
         ctxt = context.get_admin_context()
@@ -104,6 +104,9 @@ class API(base.Base):
         if volume['status'] != "available":
             msg = _('Volume to be backed up must be available')
             raise exception.InvalidVolume(reason=msg)
+        if not self._is_backup_service_enabled(volume):
+            raise exception.ServiceNotFound(service_id='cinder-backup')
+
         self.db.volume_update(context, volume_id, {'status': 'backing-up'})
 
         options = {'user_id': context.user_id,
@@ -119,8 +122,6 @@ class API(base.Base):
                    'host': volume['host'], }
 
         backup = self.db.backup_create(context, options)
-        if not self._check_backup_service(volume):
-            raise exception.ServiceNotFound(service_id='cinder-backup')
 
         #TODO(DuncanT): In future, when we have a generic local attach,
         #               this can go via the scheduler, which enables

cinder/tests/api/contrib/test_backups.py

@@ -496,7 +496,10 @@ class BackupsAPITestCase(test.TestCase):
         self.assertEqual(res_dict['computeFault']['message'],
                          'Service cinder-backup could not be found.')
 
-    def test_check_backup_service(self):
+        volume = self.volume_api.get(context.get_admin_context(), volume_id)
+        self.assertEqual(volume['status'], 'available')
+
+    def test_is_backup_service_enabled(self):
         def empty_service(ctxt, topic):
             return []
 
@@ -532,27 +535,33 @@ class BackupsAPITestCase(test.TestCase):
 
         #test empty service
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', empty_service)
-        self.assertEqual(self.backup_api._check_backup_service(volume), False)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         False)
 
         #test host not match service
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', host_not_match)
-        self.assertEqual(self.backup_api._check_backup_service(volume), False)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         False)
 
         #test az not match service
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', az_not_match)
-        self.assertEqual(self.backup_api._check_backup_service(volume), False)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         False)
 
         #test disabled service
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', disabled_service)
-        self.assertEqual(self.backup_api._check_backup_service(volume), False)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         False)
 
         #test dead service
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', dead_service)
-        self.assertEqual(self.backup_api._check_backup_service(volume), False)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         False)
 
         #test multi services and the last service matches
         self.stubs.Set(cinder.db, 'service_get_all_by_topic', multi_services)
-        self.assertEqual(self.backup_api._check_backup_service(volume), True)
+        self.assertEqual(self.backup_api._is_backup_service_enabled(volume),
+                         True)
 
     def test_delete_backup_available(self):
         backup_id = self._create_backup(status='available')