1. Add _get and _delete method to BackupAdminController.
2. Add force=False arg to delete method in backup api.
3. If force=True, then allow the deletion of backup, which
can be in any state.
4. For chunkeddriver based backup, check the status of backup
from DB in backup loop, if it has changed to deleting, set the
flag is_backup_canceled and clear the data to avoid leaving orphan
object in the object store.
5. Add the flag: support_force_delete=False in BackupDriver. That
indicates if the backup driver supports force deletion.
It should be set to True if the driver that inherits from BackupDriver
supports the force deletion function.
6. If the backup driver doesn't support this feature, and it will return
405 when calling API of force delete.
DocImpact
APIImpact
Add os-force_delete in request body:
*POST /v2/{tenant_id}/backups/{id}/action
{
"os-force_delete": {}
}
Implements: blueprint support-force-delete-backup
Change-Id: Ia4d36520d2a092730cd1be56e338349e383f9e6b
'error'
])
+ def _get(self, *args, **kwargs):
+ return self.backup_api.get(*args, **kwargs)
+
+ def _delete(self, *args, **kwargs):
+ return self.backup_api.delete(*args, **kwargs)
+
@wsgi.action('os-reset_status')
def _reset_status(self, req, id, body):
"""Reset status on the resource."""
def delete(self, req, id):
"""Delete a backup."""
- LOG.debug('delete called for member %s', id)
+ LOG.debug('Delete called for member %s.', id)
context = req.environ['cinder.context']
LOG.info(_LI('Delete backup with id: %s'), id, context=context)
try:
- self.backup_api.delete(context, id)
+ backup = self.backup_api.get(context, id)
+ self.backup_api.delete(context, backup)
except exception.BackupNotFound as error:
raise exc.HTTPNotFound(explanation=error.msg)
except exception.InvalidBackup as error:
check_policy(context, 'get')
return objects.Backup.get_by_id(context, backup_id)
- def delete(self, context, backup_id):
- """Make the RPC call to delete a volume backup."""
+ def _check_support_to_force_delete(self, context, backup_host):
+ result = self.backup_rpcapi.check_support_to_force_delete(context,
+ backup_host)
+ return result
+
+ def delete(self, context, backup, force=False):
+ """Make the RPC call to delete a volume backup.
+
+ Call backup manager to execute backup delete or force delete operation.
+ :param context: running context
+ :param backup: the dict of backup that is got from DB.
+ :param force: indicate force delete or not
+ :raises: InvalidBackup
+ :raises: BackupDriverException
+ """
check_policy(context, 'delete')
- backup = self.get(context, backup_id)
- if backup['status'] not in ['available', 'error']:
+ if not force and backup.status not in ['available', 'error']:
msg = _('Backup status must be available or error')
raise exception.InvalidBackup(reason=msg)
+ if force and not self._check_support_to_force_delete(context,
+ backup.host):
+ msg = _('force delete')
+ raise exception.NotSupportedOperation(operation=msg)
# Don't allow backup to be deleted if there are incremental
# backups dependent on it.
- deltas = self.get_all(context, {'parent_id': backup['id']})
+ deltas = self.get_all(context, {'parent_id': backup.id})
if deltas and len(deltas):
msg = _('Incremental backups exist for this backup.')
raise exception.InvalidBackup(reason=msg)
self.backup_compression_algorithm = CONF.backup_compression_algorithm
self.compressor = \
self._get_compressor(CONF.backup_compression_algorithm)
+ self.support_force_delete = True
# To create your own "chunked" backup driver, implement the following
# abstract methods.
sha256_list = object_sha256['sha256s']
shaindex = 0
+ is_backup_canceled = False
while True:
+ # First of all, we check the status of this backup. If it
+ # has been changed to delete or has been deleted, we cancel the
+ # backup process to do forcing delete.
+ backup = objects.Backup.get_by_id(self.context, backup.id)
+ if 'deleting' == backup.status or 'deleted' == backup.status:
+ is_backup_canceled = True
+ # To avoid the chunk left when deletion complete, need to
+ # clean up the object of chunk again.
+ self.delete(backup)
+ LOG.debug('Cancel the backup process of %s.', backup.id)
+ break
data_offset = volume_file.tell()
data = volume_file.read(self.chunk_size_bytes)
if data == b'':
# Stop the timer.
timer.stop()
+ # If backup has been cancelled we have nothing more to do
+ # but timer.stop().
+ if is_backup_canceled:
+ return
# All the data have been sent, the backup_percent reaches 100.
self._send_progress_end(self.context, backup, object_meta)
super(BackupDriver, self).__init__(db_driver)
self.context = context
self.backup_meta_api = BackupMetadataAPI(context, db_driver)
+ # This flag indicates if backup driver supports force
+ # deletion. So it should be set to True if the driver that inherits
+ # from BackupDriver supports the force deletion function.
+ self.support_force_delete = False
def get_metadata(self, volume_id):
return self.backup_meta_api.get(volume_id)
notifier = rpc.get_notifier('backupStatusUpdate')
notifier.info(context, "backups.reset_status.end",
notifier_info)
+
+ def check_support_to_force_delete(self, context):
+ """Check if the backup driver supports force delete operation.
+
+ :param context: running context
+ """
+ backup_service = self.service.get_backup_driver(context)
+ return backup_service.support_force_delete
'host': backup.host})
cctxt = self.client.prepare(server=backup.host)
return cctxt.cast(ctxt, 'reset_status', backup=backup, status=status)
+
+ def check_support_to_force_delete(self, ctxt, host):
+ LOG.debug("Check if backup driver supports force delete "
+ "on host %(host)s.", {'host': host})
+ cctxt = self.client.prepare(server=host)
+ return cctxt.call(ctxt, 'check_support_to_force_delete')
class MetadataAbsent(CinderException):
message = _("There is no metadata in DB object.")
+
+
+class NotSupportedOperation(Invalid):
+ message = _("Operation not supported: %(operation)s.")
+ code = 405
from cinder import exception
from cinder import objects
from cinder import test
+from cinder.tests.unit.api.contrib import test_backups
from cinder.tests.unit.api import fakes
from cinder.tests.unit.api.v2 import stubs
from cinder.tests.unit import cast_as_call
self.assertRaises(exc.HTTPBadRequest,
vac.validate_update,
{'status': 'creating'})
+
+ @mock.patch('cinder.backup.api.API._check_support_to_force_delete')
+ def _force_delete_backup_util(self, test_status, mock_check_support):
+ # admin context
+ ctx = context.RequestContext('admin', 'fake', True)
+ mock_check_support.return_value = True
+ # current status is dependent on argument: test_status.
+ id = test_backups.BackupsAPITestCase._create_backup(status=test_status)
+ req = webob.Request.blank('/v2/fake/backups/%s/action' % id)
+ req.method = 'POST'
+ req.headers['Content-Type'] = 'application/json'
+ req.body = jsonutils.dumps({'os-force_delete': {}})
+ req.environ['cinder.context'] = ctx
+ res = req.get_response(app())
+
+ self.assertEqual(202, res.status_int)
+ self.assertEqual('deleting',
+ test_backups.BackupsAPITestCase.
+ _get_backup_attrib(id, 'status'))
+ db.backup_destroy(context.get_admin_context(), id)
+
+ def test_delete_backup_force_when_creating(self):
+ self._force_delete_backup_util('creating')
+
+ def test_delete_backup_force_when_deleting(self):
+ self._force_delete_backup_util('deleting')
+
+ def test_delete_backup_force_when_restoring(self):
+ self._force_delete_backup_util('restoring')
+
+ def test_delete_backup_force_when_available(self):
+ self._force_delete_backup_util('available')
+
+ def test_delete_backup_force_when_error(self):
+ self._force_delete_backup_util('error')
+
+ def test_delete_backup_force_when_error_deleting(self):
+ self._force_delete_backup_util('error_deleting')
+
+ @mock.patch('cinder.backup.rpcapi.BackupAPI.check_support_to_force_delete',
+ return_value=False)
+ def test_delete_backup_force_when_not_supported(self, mock_check_support):
+ # admin context
+ ctx = context.RequestContext('admin', 'fake', True)
+ self.override_config('backup_driver', 'cinder.backup.drivers.ceph')
+ id = test_backups.BackupsAPITestCase._create_backup()
+ req = webob.Request.blank('/v2/fake/backups/%s/action' % id)
+ req.method = 'POST'
+ req.headers['Content-Type'] = 'application/json'
+ req.body = jsonutils.dumps({'os-force_delete': {}})
+ req.environ['cinder.context'] = ctx
+ res = req.get_response(app())
+ self.assertEqual(405, res.status_int)
self.assertEqual("Missing required element 'backup-record' in "
"request body.",
res_dict['badRequest']['message'])
+
+ @mock.patch('cinder.backup.rpcapi.BackupAPI.check_support_to_force_delete',
+ return_value=False)
+ def test_force_delete_with_not_supported_operation(self,
+ mock_check_support):
+ backup_id = self._create_backup(status='available')
+ backup = self.backup_api.get(self.context, backup_id)
+ self.assertRaises(exception.NotSupportedOperation,
+ self.backup_api.delete, self.context, backup, True)
"volume_extension:volume_admin_actions:reset_status": "rule:admin_api",
"volume_extension:snapshot_admin_actions:reset_status": "rule:admin_api",
"volume_extension:backup_admin_actions:reset_status": "rule:admin_api",
+ "volume_extension:backup_admin_actions:force_delete": "rule:admin_api",
"volume_extension:volume_admin_actions:force_delete": "rule:admin_api",
"volume_extension:snapshot_admin_actions:force_delete": "rule:admin_api",
"volume_extension:volume_admin_actions:force_detach": "rule:admin_api",
backup = db.backup_get(self.ctxt, imported_record.id)
self.assertEqual(backup['status'], 'error')
+ def test_not_supported_driver_to_force_delete(self):
+ """Test force delete check method for not supported drivers."""
+ self.override_config('backup_driver', 'cinder.backup.drivers.ceph')
+ self.backup_mgr = importutils.import_object(CONF.backup_manager)
+ result = self.backup_mgr.check_support_to_force_delete(self.ctxt)
+ self.assertFalse(result)
+
+ @mock.patch('cinder.backup.drivers.nfs.NFSBackupDriver.'
+ '_init_backup_repo_path', return_value=None)
+ @mock.patch('cinder.backup.drivers.nfs.NFSBackupDriver.'
+ '_check_configuration', return_value=None)
+ def test_check_support_to_force_delete(self, mock_check_configuration,
+ mock_init_backup_repo_path):
+ """Test force delete check method for supported drivers."""
+ self.override_config('backup_driver', 'cinder.backup.drivers.nfs')
+ self.backup_mgr = importutils.import_object(CONF.backup_manager)
+ result = self.backup_mgr.check_support_to_force_delete(self.ctxt)
+ self.assertTrue(result)
+
class BackupTestCaseWithVerify(BaseBackupTest):
"""Test Case for backups."""
"volume_extension:volume_admin_actions:force_delete": "rule:admin_api",
"volume_extension:volume_admin_actions:force_detach": "rule:admin_api",
"volume_extension:snapshot_admin_actions:force_delete": "rule:admin_api",
+ "volume_extension:backup_admin_actions:force_delete": "rule:admin_api",
"volume_extension:volume_admin_actions:migrate_volume": "rule:admin_api",
"volume_extension:volume_admin_actions:migrate_volume_completion": "rule:admin_api",
Code Review / openstack-build / cinder-build.git / commitdiff