clear_volume was almost the same in LVM and BlockDevice drivers.
So it has been moved to utils.
Change-Id: Iadb6b8d01cf500109bb48b0d2c817109918519e0
from cinder.tests.api import fakes
from cinder.tests.api.v2 import stubs
from cinder.volume import api as volume_api
+from cinder.volume import utils as volutils
CONF = cfg.CONF
def test_force_delete_snapshot(self):
self.stubs.Set(os.path, 'exists', lambda x: True)
+ self.stubs.Set(volutils, 'clear_volume',
+ lambda a, b, volume_clear=CONF.volume_clear,
+ volume_clear_size=CONF.volume_clear_size: None)
# admin context
ctx = context.RequestContext('admin', 'fake', True)
# current status is creating
TEST_VOLUME2 = {'provider_location': None}
self.mox.StubOutWithMock(self.drv, 'local_path')
self.drv.local_path(TEST_VOLUME2).AndReturn(None)
- self.mox.StubOutWithMock(self.drv, 'clear_volume')
+ self.mox.StubOutWithMock(volutils, 'clear_volume')
self.mox.ReplayAll()
self.drv.delete_volume(TEST_VOLUME2)
path = self.drv.local_path(TEST_VOLUME1).AndReturn('/dev/loop1')
self.mox.StubOutWithMock(os.path, 'exists')
os.path.exists(path).AndReturn(True)
- self.mox.StubOutWithMock(self.drv, 'clear_volume')
- self.drv.clear_volume(TEST_VOLUME1)
+ self.mox.StubOutWithMock(volutils, 'clear_volume')
+ self.mox.StubOutWithMock(self.drv, '_get_device_size')
+ size = self.drv._get_device_size(path).AndReturn(1024)
+ volutils.clear_volume(size, path,
+ volume_clear=mox.IgnoreArg(),
+ volume_clear_size=mox.IgnoreArg())
self.mox.ReplayAll()
self.drv.delete_volume(TEST_VOLUME1)
TEST_VOLUME2 = {'provider_location': '1 2 3 /dev/loop0'}
self.mox.StubOutWithMock(self.drv, 'local_path')
self.drv.local_path(TEST_VOLUME2).AndReturn('/dev/loop0')
- self.mox.StubOutWithMock(self.drv, 'clear_volume')
+ self.mox.StubOutWithMock(volutils, 'clear_volume')
self.mox.ReplayAll()
self.drv.delete_volume(TEST_VOLUME2)
import contextlib
import datetime
import os
-import re
import shutil
import socket
import tempfile
self.volume_params = {
'status': 'creating',
'host': CONF.host,
- 'size': 0}
+ 'size': 1}
self.stubs.Set(iscsi.TgtAdm, '_get_target', self.fake_get_target)
self.stubs.Set(brick_lvm.LVM,
'get_all_volume_groups',
class VolumeTestCase(BaseVolumeTestCase):
+
+ def setUp(self):
+ super(VolumeTestCase, self).setUp()
+ self.stubs.Set(volutils, 'clear_volume',
+ lambda a, b, volume_clear=mox.IgnoreArg(),
+ volume_clear_size=mox.IgnoreArg(),
+ lvm_type=mox.IgnoreArg(): None)
+
def test_init_host_clears_downloads(self):
"""Test that init_host will unwedge a volume stuck in downloading."""
volume = tests_utils.create_volume(self.context, status='downloading',
'snapshot_id': None,
'user_id': 'fake',
'launched_at': 'DONTCARE',
- 'size': 0,
+ 'size': 1,
}
self.assertDictMatch(msg['payload'], expected)
msg = test_notifier.NOTIFICATIONS[1]
result_dict['snapshot_metadata'][0].key:
result_dict['snapshot_metadata'][0].value}
self.assertEqual(result_meta, test_meta)
-
self.volume.delete_snapshot(self.context, snapshot_id)
self.assertRaises(exception.NotFound,
db.snapshot_get,
class LVMVolumeDriverTestCase(DriverTestCase):
"""Test case for VolumeDriver"""
driver_name = "cinder.volume.drivers.lvm.LVMVolumeDriver"
+ FAKE_VOLUME = {'name': 'test1',
+ 'id': 'test1'}
- def test_clear_volume(self):
+ def test_delete_volume_invalid_parameter(self):
configuration = conf.Configuration(fake_opt, 'fake_group')
configuration.volume_clear = 'zero'
configuration.volume_clear_size = 0
lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
- self.mox.StubOutWithMock(volutils, 'copy_volume')
self.mox.StubOutWithMock(os.path, 'exists')
- self.mox.StubOutWithMock(utils, 'execute')
-
- fake_volume = {'name': 'test1',
- 'volume_name': 'test1',
- 'id': 'test1'}
-
- os.path.exists(mox.IgnoreArg()).AndReturn(True)
- volutils.copy_volume('/dev/zero', mox.IgnoreArg(), 123 * 1024,
- mox.IgnoreArg(), execute=lvm_driver._execute,
- sync=True)
-
- os.path.exists(mox.IgnoreArg()).AndReturn(True)
- volutils.copy_volume('/dev/zero', mox.IgnoreArg(), 123 * 1024,
- mox.IgnoreArg(), execute=lvm_driver._execute,
- sync=True)
os.path.exists(mox.IgnoreArg()).AndReturn(True)
self.mox.ReplayAll()
- # Test volume has 'size' field
- volume = dict(fake_volume, size=123)
- lvm_driver.clear_volume(volume)
-
- # Test volume has 'volume_size' field
- volume = dict(fake_volume, volume_size=123)
- lvm_driver.clear_volume(volume)
-
# Test volume without 'size' field and 'volume_size' field
- volume = dict(fake_volume)
self.assertRaises(exception.InvalidParameterValue,
- lvm_driver.clear_volume,
- volume)
+ lvm_driver._delete_volume,
+ self.FAKE_VOLUME)
- def test_clear_volume_badopt(self):
+ def test_delete_volume_bad_path(self):
configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.volume_clear = 'non_existent_volume_clearer'
+ configuration.volume_clear = 'zero'
configuration.volume_clear_size = 0
+ volume = dict(self.FAKE_VOLUME, size=1)
lvm_driver = lvm.LVMVolumeDriver(configuration=configuration)
- self.mox.StubOutWithMock(volutils, 'copy_volume')
- self.mox.StubOutWithMock(os.path, 'exists')
-
- fake_volume = {'name': 'test1',
- 'volume_name': 'test1',
- 'id': 'test1',
- 'size': 123}
-
- os.path.exists(mox.IgnoreArg()).AndReturn(True)
+ self.mox.StubOutWithMock(os.path, 'exists')
+ os.path.exists(mox.IgnoreArg()).AndReturn(False)
self.mox.ReplayAll()
- volume = dict(fake_volume)
- self.assertRaises(exception.InvalidConfigurationValue,
- lvm_driver.clear_volume,
- volume)
+ self.assertRaises(exception.VolumeBackendAPIException,
+ lvm_driver._delete_volume, volume)
- def test_clear_volume_thinlvm_snap(self):
+ def test_delete_volume_thinlvm_snap(self):
configuration = conf.Configuration(fake_opt, 'fake_group')
configuration.volume_clear = 'zero'
configuration.volume_clear_size = 0
configuration.lvm_type = 'thin'
- lvm_driver = lvm.LVMISCSIDriver(configuration=configuration)
+ lvm_driver = lvm.LVMISCSIDriver(configuration=configuration,
+ vg_obj=mox.MockAnything())
# Ensures that copy_volume is not called for ThinLVM
self.mox.StubOutWithMock(volutils, 'copy_volume')
+ self.mox.StubOutWithMock(volutils, 'clear_volume')
self.mox.StubOutWithMock(lvm_driver, '_execute')
+ self.mox.ReplayAll()
uuid = '00000000-0000-0000-0000-c3aa7ee01536'
'id': uuid,
'size': 123}
- lvm_driver.clear_volume(fake_snapshot, is_snapshot=True)
-
- def test_clear_volume_lvm_snap(self):
- self.stubs.Set(os.path, 'exists', lambda x: True)
- configuration = conf.Configuration(fake_opt, 'fake_group')
- configuration.volume_clear = 'zero'
- configuration.volume_clear_size = 0
- lvm_driver = lvm.LVMISCSIDriver(configuration=configuration)
-
- uuid = '00000000-0000-0000-0000-90ed32cdeed3'
- name = 'snapshot-' + uuid
- mangle_name = '_' + re.sub(r'-', r'--', name)
-
- def fake_copy_volume(srcstr, deststr, size, blocksize, **kwargs):
- self.assertEqual(deststr,
- '/dev/mapper/cinder--volumes-%s-cow' %
- mangle_name)
- return True
-
- self.stubs.Set(volutils, 'copy_volume', fake_copy_volume)
-
- fake_snapshot = {'name': 'snapshot-' + uuid,
- 'id': uuid,
- 'size': 123}
-
- lvm_driver.clear_volume(fake_snapshot, is_snapshot=True)
+ lvm_driver._delete_volume(fake_snapshot, is_snapshot=True)
class ISCSITestCase(DriverTestCase):
"""Tests For miscellaneous util methods used with volume."""
+import os
+import re
from oslo.config import cfg
from cinder import context
from cinder import db
+from cinder import exception
from cinder.openstack.common import importutils
from cinder.openstack.common import log as logging
from cinder.openstack.common.notifier import api as notifier_api
from cinder.openstack.common.notifier import test_notifier
from cinder import test
+from cinder import utils
from cinder.volume import utils as volume_utils
bs, count = volume_utils._calculate_count(1024, 'ABM')
self.assertEqual(bs, '1M')
self.assertEqual(count, 1024)
+
+
+class ClearVolumeTestCase(test.TestCase):
+
+ def test_clear_volume(self):
+ CONF.volume_clear = 'zero'
+ CONF.volume_clear_size = 0
+ CONF.volume_dd_blocksize = '1M'
+ self.mox.StubOutWithMock(volume_utils, 'copy_volume')
+ volume_utils.copy_volume("/dev/zero", "volume_path", 1024,
+ CONF.volume_dd_blocksize, sync=True,
+ execute=utils.execute)
+ self.mox.ReplayAll()
+ volume_utils.clear_volume(1024, "volume_path")
+
+ def test_clear_volume_zero_and_shred(self):
+ CONF.volume_clear = 'zero'
+ CONF.volume_clear_size = 1
+ clear_cmd = ['shred', '-n0', '-z', '-s1MiB', "volume_path"]
+ self.mox.StubOutWithMock(utils, "execute")
+ utils.execute(*clear_cmd, run_as_root=True)
+ self.mox.ReplayAll()
+ volume_utils.clear_volume(1024, "volume_path")
+
+ def test_clear_volume_shred(self):
+ CONF.volume_clear = 'shred'
+ CONF.volume_clear_size = 1
+ clear_cmd = ['shred', '-n3', '-s1MiB', "volume_path"]
+ self.mox.StubOutWithMock(utils, "execute")
+ utils.execute(*clear_cmd, run_as_root=True)
+ self.mox.ReplayAll()
+ volume_utils.clear_volume(1024, "volume_path")
+
+ def test_clear_volume_shred_not_clear_size(self):
+ CONF.volume_clear = 'shred'
+ CONF.volume_clear_size = None
+ clear_cmd = ['shred', '-n3', "volume_path"]
+ self.mox.StubOutWithMock(utils, "execute")
+ utils.execute(*clear_cmd, run_as_root=True)
+ self.mox.ReplayAll()
+ volume_utils.clear_volume(1024, "volume_path")
+
+ def test_clear_volume_invalid_opt(self):
+ CONF.volume_clear = 'non_existent_volume_clearer'
+ CONF.volume_clear_size = 0
+ self.mox.StubOutWithMock(volume_utils, 'copy_volume')
+
+ self.mox.ReplayAll()
+
+ self.assertRaises(exception.InvalidConfigurationValue,
+ volume_utils.clear_volume,
+ 1024, "volume_path")
+
+ def test_clear_volume_lvm_snap(self):
+ self.stubs.Set(os.path, 'exists', lambda x: True)
+ CONF.volume_clear = 'zero'
+ CONF.volume_clear_size = 0
+
+ uuid = '00000000-0000-0000-0000-90ed32cdeed3'
+ name = 'snapshot-' + uuid
+ mangle_name = '_' + re.sub(r'-', r'--', name)
+ vol_path = '/dev/mapper/cinder--volumes-%s-cow' % mangle_name
+
+ def fake_copy_volume(srcstr, deststr, size, blocksize, **kwargs):
+ self.assertEqual(deststr, vol_path)
+ return True
+
+ self.stubs.Set(volume_utils, 'copy_volume', fake_copy_volume)
+ volume_utils.clear_volume(123, vol_path)
if not dev_path or dev_path not in \
self.configuration.available_devices:
return
- if os.path.exists(dev_path):
- self.clear_volume(volume)
+ if os.path.exists(dev_path) and \
+ self.configuration.volume_clear != 'none':
+ volutils.clear_volume(
+ self._get_device_size(dev_path), dev_path,
+ volume_clear=self.configuration.volume_clear,
+ volume_clear_size=self.configuration.volume_clear_size)
def local_path(self, volume):
if volume['provider_location']:
else:
return None
- def clear_volume(self, volume):
- """unprovision old volumes to prevent data leaking between users."""
- vol_path = self.local_path(volume)
- clear_size = self.configuration.volume_clear_size
- size_in_m = self._get_device_size(vol_path)
-
- if self.configuration.volume_clear == 'none':
- return
-
- LOG.info(_("Performing secure delete on volume: %s") % volume['id'])
-
- if self.configuration.volume_clear == 'zero':
- if clear_size == 0:
- return volutils.copy_volume(
- '/dev/zero', vol_path, size_in_m,
- self.configuration.volume_dd_blocksize,
- sync=True, execute=self._execute)
- else:
- clear_cmd = ['shred', '-n0', '-z', '-s%dMiB' % clear_size]
- elif self.configuration.volume_clear == 'shred':
- clear_cmd = ['shred', '-n3']
- if clear_size:
- clear_cmd.append('-s%dMiB' % clear_size)
- else:
- LOG.error(_("Error unrecognized volume_clear option: %s"),
- self.configuration.volume_clear)
- return
-
- clear_cmd.append(vol_path)
- self._execute(*clear_cmd, run_as_root=True)
-
def copy_image_to_volume(self, context, volume, image_service, image_id):
"""Fetch the image from image_service and write it to the volume."""
image_utils.fetch_to_raw(context,
# zero out old volumes to prevent data leaking between users
# TODO(ja): reclaiming space should be done lazy and low priority
- self.clear_volume(volume, is_snapshot)
+ if not self.configuration.lvm_type == 'thin' and \
+ self.configuration.volume_clear != 'none':
+ if is_snapshot:
+ # if the volume to be cleared is a snapshot of another volume
+ # we need to clear out the volume using the -cow instead of the
+ # directly volume path. We need to skip this if we are using
+ # thin provisioned LVs.
+ # bug# lp1191812
+ dev_path = self.local_path(volume) + "-cow"
+ else:
+ dev_path = self.local_path(volume)
+
+ # TODO(jdg): Maybe we could optimize this for snaps by looking at
+ # the cow table and only overwriting what's necessary?
+ # for now we're still skipping on snaps due to hang issue
+ if not os.path.exists(dev_path):
+ msg = (_('Volume device file path %s does not exist.')
+ % dev_path)
+ LOG.error(msg)
+ raise exception.VolumeBackendAPIException(data=msg)
+
+ size_in_g = volume.get('size', volume.get('volume_size', None))
+ if size_in_g is None:
+ msg = (_("Size for volume: %s not found, "
+ "cannot secure delete.") % volume['id'])
+ LOG.error(msg)
+ raise exception.InvalidParameterValue(msg)
+ vol_size = size_in_g * 1024
+
+ volutils.clear_volume(
+ vol_size, dev_path,
+ volume_clear=self.configuration.volume_clear,
+ volume_clear_size=self.configuration.volume_clear_size)
name = volume['name']
if is_snapshot:
name = self._escape_snapshot(volume['name'])
self._delete_volume(volume)
- def clear_volume(self, volume, is_snapshot=False):
- """unprovision old volumes to prevent data leaking between users."""
-
- # NOTE(jdg): Don't write the blocks of thin provisioned
- # volumes
- if self.configuration.volume_clear == 'none' or \
- self.configuration.lvm_type == 'thin':
- return
-
- if is_snapshot:
- # if the volume to be cleared is a snapshot of another volume
- # we need to clear out the volume using the -cow instead of the
- # directly volume path. We need to skip this if we are using
- # thin provisioned LVs.
- # bug# lp1191812
- dev_path = self.local_path(volume) + "-cow"
- else:
- dev_path = self.local_path(volume)
-
- if not os.path.exists(dev_path):
- msg = (_('Volume device file path %s does not exist.') % dev_path)
- LOG.error(msg)
- raise exception.VolumeBackendAPIException(data=msg)
-
- size_in_g = volume.get('size', volume.get('volume_size', None))
- if size_in_g is None:
- msg = (_("Size for volume: %s not found, "
- "cannot secure delete.") % volume['id'])
- LOG.error(msg)
- raise exception.InvalidParameterValue(msg)
- size_in_m = self.configuration.volume_clear_size
-
- LOG.info(_("Performing secure delete on volume: %s") % volume['id'])
-
- if self.configuration.volume_clear == 'zero':
- if size_in_m == 0:
- return volutils.copy_volume(
- '/dev/zero',
- dev_path, size_in_g * 1024,
- self.configuration.volume_dd_blocksize,
- sync=True,
- execute=self._execute)
- else:
- clear_cmd = ['shred', '-n0', '-z', '-s%dMiB' % size_in_m]
- elif self.configuration.volume_clear == 'shred':
- clear_cmd = ['shred', '-n3']
- if size_in_m:
- clear_cmd.append('-s%dMiB' % size_in_m)
- else:
- raise exception.InvalidConfigurationValue(
- option='volume_clear',
- value=self.configuration.volume_clear)
-
- clear_cmd.append(dev_path)
- self._execute(*clear_cmd, run_as_root=True)
-
def create_snapshot(self, snapshot):
"""Creates a snapshot."""
from oslo.config import cfg
from cinder.brick.local_dev import lvm as brick_lvm
+from cinder import exception
from cinder.openstack.common import log as logging
from cinder.openstack.common.notifier import api as notifier_api
from cinder.openstack.common import processutils
*extra_flags, run_as_root=True)
+def clear_volume(volume_size, volume_path, volume_clear=None,
+ volume_clear_size=None):
+ """Unprovision old volumes to prevent data leaking between users."""
+ if volume_clear is None:
+ volume_clear = CONF.volume_clear
+
+ if volume_clear_size is None:
+ volume_clear_size = CONF.volume_clear_size
+
+ LOG.info(_("Performing secure delete on volume: %s") % volume_path)
+
+ if volume_clear == 'zero':
+ if volume_clear_size == 0:
+ return copy_volume('/dev/zero', volume_path, volume_size,
+ CONF.volume_dd_blocksize,
+ sync=True, execute=utils.execute)
+ else:
+ clear_cmd = ['shred', '-n0', '-z', '-s%dMiB' % volume_clear_size]
+ elif volume_clear == 'shred':
+ clear_cmd = ['shred', '-n3']
+ if volume_clear_size:
+ clear_cmd.append('-s%dMiB' % volume_clear_size)
+ else:
+ raise exception.InvalidConfigurationValue(
+ option='volume_clear',
+ value=volume_clear)
+
+ clear_cmd.append(volume_path)
+ utils.execute(*clear_cmd, run_as_root=True)
+
+
def supports_thin_provisioning():
return brick_lvm.LVM.supports_thin_provisioning(
utils.get_root_helper())
Code Review / openstack-build / cinder-build.git / commitdiff