openstack
/
cinder
Code Issues Proposed changes

Prevent temporary volume from being deleted accidentally 

A temporary volume can be deleted while it is in use by DELETE API
accidentally because its status is `available`. To avoid this deletion,
this fix sets a value which doesn't accept deletion to volume status of
a temporary volume. When a temporary volume is used for backing up,
`backing-up` is set. When a temporary volume is used for reverting a
snapshot, `in-use` is set because the volume is attached by a host.

Closes-Bug: #1970768
Change-Id: Ib6a2e4d68e532b91161df5245c17ce815f12f935
This commit is contained in:
Hironori Shiina 2022-01-28 21:38:03 -05:00
parent d5b21ab1ec
commit 53c13891b3
6 changed files with 32 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
cinder/tests/unit/backup/test_backup.py
View File

@ -1019,9 +1019,9 @@ class BackupTestCase(BaseBackupTest):
'fake_provider_id'}
temp_vol = volume_manager.driver._create_temp_cloned_volume(
self.ctxt, vol)
self.ctxt, vol, status=fields.VolumeStatus.BACKING_UP)
self.assertEqual('available', temp_vol['status'])
self.assertEqual(fields.VolumeStatus.BACKING_UP, temp_vol['status'])
self.assertEqual('fake_provider_id', temp_vol['provider_id'])
@mock.patch.object(fake_driver.FakeLoggingVolumeDriver,
@ -1038,9 +1038,9 @@ class BackupTestCase(BaseBackupTest):
'fake_provider_id'}
temp_vol = volume_manager.driver._create_temp_volume_from_snapshot(
self.ctxt, vol, snap)
self.ctxt, vol, snap, status=fields.VolumeStatus.BACKING_UP)
self.assertEqual('available', temp_vol['status'])
self.assertEqual(fields.VolumeStatus.BACKING_UP, temp_vol['status'])
self.assertEqual('fake_provider_id', temp_vol['provider_id'])
@mock.patch('cinder.volume.volume_utils.notify_about_backup_usage')

11
cinder/tests/unit/volume/test_driver.py
View File

@ -209,8 +209,8 @@ class GenericVolumeDriverTestCase(BaseDriverTestCase):
vol = tests_utils.create_volume(self.context,
status='backing-up')
cloned_vol = self.volume.driver._create_temp_cloned_volume(
self.context, vol)
self.assertEqual('available', cloned_vol.status)
self.context, vol, status=fields.VolumeStatus.BACKING_UP)
self.assertEqual(fields.VolumeStatus.BACKING_UP, cloned_vol.status)
def test_get_backup_device_available(self):
vol = tests_utils.create_volume(self.context)
@ -232,8 +232,9 @@ class GenericVolumeDriverTestCase(BaseDriverTestCase):
status='backing-up',
previous_status='in-use')
admin_meta = {'temporary': 'True'}
temp_vol = tests_utils.create_volume(self.context,
admin_metadata=admin_meta)
temp_vol = tests_utils.create_volume(
self.context, status=fields.VolumeStatus.BACKING_UP,
admin_metadata=admin_meta)
self.context.user_id = fake.USER_ID
self.context.project_id = fake.PROJECT_ID
backup_obj = tests_utils.create_backup(self.context,
@ -249,6 +250,8 @@ class GenericVolumeDriverTestCase(BaseDriverTestCase):
self.assertFalse(is_snapshot)
backup_obj.refresh()
self.assertEqual(temp_vol.id, backup_obj.temp_volume_id)
mock_create_temp.assert_called_once_with(
self.context, vol, status=fields.VolumeStatus.BACKING_UP)
def test_create_temp_volume_from_snapshot(self):
volume_dict = {'id': fake.SNAPSHOT_ID,

4
cinder/tests/unit/volume/test_volume.py
View File

@ -2247,8 +2247,8 @@ class VolumeTestCase(base.BaseVolumeTestCase):
'delete_volume') as mock_driver_delete, \
mock.patch.object(
self.volume, '_copy_volume_data') as mock_copy:
temp_volume = tests_utils.create_volume(self.context,
status='available')
temp_volume = tests_utils.create_volume(
self.context, status=fields.VolumeStatus.IN_USE)
mock_copy.side_effect = [exception.VolumeDriverException('error')]
mock_temp.return_value = temp_volume

17
cinder/volume/driver.py
View File

@ -1219,7 +1219,8 @@ class BaseVD(object, metaclass=abc.ABCMeta):
# then clean up the temp volume.
if snapshot:
temp_vol_ref = self._create_temp_volume_from_snapshot(
context, volume, snapshot)
context, volume, snapshot,
status=fields.VolumeStatus.BACKING_UP)
backup.temp_volume_id = temp_vol_ref.id
backup.save()
device_to_backup = temp_vol_ref
@ -1232,7 +1233,7 @@ class BaseVD(object, metaclass=abc.ABCMeta):
previous_status = volume.get('previous_status')
if previous_status == "in-use":
temp_vol_ref = self._create_temp_cloned_volume(
context, volume)
context, volume, status=fields.VolumeStatus.BACKING_UP)
backup.temp_volume_id = temp_vol_ref.id
backup.save()
device_to_backup = temp_vol_ref
@ -1334,7 +1335,8 @@ class BaseVD(object, metaclass=abc.ABCMeta):
temp_vol_ref.create()
return temp_vol_ref
def _create_temp_cloned_volume(self, context, volume):
def _create_temp_cloned_volume(self, context, volume,
status=fields.VolumeStatus.AVAILABLE):
temp_vol_ref = self._create_temp_volume(context, volume)
try:
model_update = self.create_cloned_volume(temp_vol_ref, volume)
@ -1344,12 +1346,13 @@ class BaseVD(object, metaclass=abc.ABCMeta):
with excutils.save_and_reraise_exception():
temp_vol_ref.destroy()
temp_vol_ref.status = 'available'
temp_vol_ref.status = status
temp_vol_ref.save()
return temp_vol_ref
def _create_temp_volume_from_snapshot(self, context, volume, snapshot,
volume_options=None):
def _create_temp_volume_from_snapshot(
self, context, volume, snapshot, volume_options=None,
status=fields.VolumeStatus.AVAILABLE):
temp_vol_ref = self._create_temp_volume(context, volume,
volume_options=volume_options)
try:
@ -1361,7 +1364,7 @@ class BaseVD(object, metaclass=abc.ABCMeta):
with excutils.save_and_reraise_exception():
temp_vol_ref.destroy()
temp_vol_ref.status = 'available'
temp_vol_ref.status = status
temp_vol_ref.save()
return temp_vol_ref

5
cinder/volume/manager.py
View File

@ -1068,7 +1068,8 @@ class VolumeManager(manager.CleanableManager,
'from snapshot %s' % snapshot.id}
ctxt = context.get_internal_tenant_context() or ctxt
temp_vol = self.driver._create_temp_volume_from_snapshot(
ctxt, volume, snapshot, volume_options=v_options)
ctxt, volume, snapshot, volume_options=v_options,
status=fields.VolumeStatus.IN_USE)
self._copy_volume_data(ctxt, temp_vol, volume)
self.driver_delete_volume(temp_vol)
temp_vol.destroy()
@ -1080,7 +1081,7 @@ class VolumeManager(manager.CleanableManager,
" %(volume)s.",
{'snapshot': snapshot.id,
'volume': volume.id})
if temp_vol and temp_vol.status == 'available':
if temp_vol and temp_vol.status == fields.VolumeStatus.IN_USE:
self.driver_delete_volume(temp_vol)
temp_vol.destroy()

6
releasenotes/notes/bug-1970768-temp-vol-delete-6586a13f08d7a5c1.yaml Normal file
View File

@ -0,0 +1,6 @@
---
fixes:
- |
`Bug #1970768 <https://bugs.launchpad.net/cinder/+bug/1970768>`_: Fixed
status of temporary volumes when creating backups and reverting to a
snapshot, preventing accidental manual deletion of those resources.