Dear autotest team,
I made an lvm test that creates a basic image, takes snapshots from it,
reverts and cleans up with the addition that it can create a volume group on
top of the ramdisk for quicker read/write ops. Some things are still hardcoded
but I can change this if the test could be of your interest.
Thanks,
Plamen
import commands
import logging
import re, os
from autotest.client import utils, test
from autotest.client.shared import error
class lvsetup(test.test):
version = 1
def _vg_ramdisk(self, vg_name):
"""Create vg on top of ram memory to speed up lv performance"""
logging.info("Creating virtual group on top of ram memory")
vg_size = "40000"
def _cleanup(vg_name, loop_device):
"""Inline cleanup function in case of test error"""
if os.path.exists("/tmp/" + vg_name):
commands.getstatusoutput("vgremove " + vg_name)
commands.getstatusoutput("pvremove " + loop_device)
commands.getstatusoutput("losetup -d " + loop_device)
commands.getstatusoutput("rm -f -R /tmp/" + vg_name +
"/virtual_hdd")
commands.getstatusoutput("umount /tmp/" + vg_name)
commands.getstatusoutput("rm -f -R /tmp/" + vg_name)
_cleanup(vg_name, "")
(status, output) = commands.getstatusoutput("mkdir /tmp/" + vg_name)
if status:
logging.warning(output)
(status, output) = commands.getstatusoutput("mount -t tmpfs tmpfs
/tmp/" + vg_name)
if status:
_cleanup(vg_name, "")
raise error.TestError(output)
cmd = "dd if=/dev/zero of=/tmp/" + vg_name + "/virtual_hdd bs=1M
count=1 seek=" + vg_size
(status, output) = commands.getstatusoutput(cmd)
if status:
_cleanup(vg_name, "")
raise error.TestError(output)
(status, output) = commands.getstatusoutput("losetup --find")
if status:
_cleanup(vg_name, "")
raise error.TestError(output)
else:
loop_device = output
(status, output) = commands.getstatusoutput("losetup " + loop_device +
" /tmp/" + vg_name + "/virtual_hdd")
if status:
_cleanup(vg_name, loop_device)
raise error.TestError(output)
(status, output) = commands.getstatusoutput("pvcreate " + loop_device)
if status:
_cleanup(vg_name, loop_device)
raise error.TestError(output)
else:
logging.info(output)
(status, output) = commands.getstatusoutput("vgcreate " + vg_name + "
" + loop_device)
if status:
_cleanup(vg_name, loop_device)
raise error.TestError(output)
else:
logging.info(output)
def _vg_check(self, vg_name):
"""Check whether provided volume group exists"""
cmd = "vgdisplay " + vg_name
(status, _) = commands.getstatusoutput(cmd)
if status:
return False
else:
logging.info("Provided logical group exists: " + vg_name)
return True
def _lv_check(self, vg_name, lv_name):
"""Check whether provided logical volume exists"""
cmd = "lvdisplay"
(status, output) = commands.getstatusoutput(cmd)
if status:
raise error.TestError("Invocation of " + cmd +
" returned status " + str(status))
# unstable approach but currently works
#lvpattern = r'LV Name\s+/dev/' + vg_name + r'/' + lv_name
lvpattern = r"LV Path\s+/dev/" + vg_name + r"/" + lv_name + "\s+"
match = re.search(lvpattern, output)
if match:
logging.info("Provided logical volume exists: /dev/" +
vg_name + "/" + lv_name)
return True
else:
return False
def _lv_remove(self, vg_name, lv_name):
"""Create a logical volume"""
logging.warning("Removing origin")
cmd = "lvremove -f " + vg_name + "/" + lv_name
(status, output) = commands.getstatusoutput(cmd)
logging.info(output)
if status:
raise error.TestError("Invocation of " + cmd +
" returned status " + str(status))
def _lv_create(self, vg_name, lv_name, lv_size):
"""Create a logical volume"""
logging.warning("Creating lv origin to take snapshot from")
cmd = ("lvcreate --size " + lv_size +
" --name " + lv_name + " " +
vg_name)
(status, output) = commands.getstatusoutput(cmd)
logging.info(output)
if status:
raise error.TestError("Invocation of " + cmd +
" returned status " + str(status))
def _lv_revert(self, vg_name, lv_snapshot_name):
"""
Revert the origin to a snapshot
Check this bug:
http://www.redhat.com/archives/linux-lvm/2011-July/msg00047.html
"""
logging.info("Reverting origin to snapshot")
cmd = ("lvconvert --merge /dev/%s/%s"
% (vg_name, lv_snapshot_name))
(status, output) = commands.getstatusoutput(cmd)
logging.info(output)
if status:
raise error.TestError("Invocation of " + cmd +
" returned status " + str(status))
def _lv_take_snapshot(self, vg_name, lv_name, lv_snapshot_name,
lv_snapshot_size):
"""Take a snapshot of the original logical volume"""
logging.info("Taking snapshot from origin")
cmd = ("lvcreate --size " + lv_snapshot_size + " --snapshot " +
" --name " + lv_snapshot_name +
" /dev/" + vg_name + "/" + lv_name)
(status, output) = commands.getstatusoutput(cmd)
logging.info(output)
if status:
raise error.TestError("Invocation of " + cmd +
" returned status " + str(status))
def run_once(self, vg_name, lv_name, lv_size, lv_snapshot_name,
lv_snapshot_size, override_flag = 0):
"""
General lv setup.
Override flag can be set to -1 for force-remove, 1 for force-create, 0
for default
"""
# if no virtual group is defined create one based on ramdisk
if not self._vg_check(vg_name):
self._vg_ramdisk(vg_name)
# if no snapshot is defined start fresh logical volume
if override_flag == 1 and self._lv_check(vg_name, lv_name):
self._lv_remove(vg_name, lv_name)
self._lv_create(vg_name, lv_name, lv_size)
elif override_flag == -1 and self._lv_check(vg_name, lv_name):
self._lv_remove(vg_name, lv_name)
else:
# perform normal check policy
if self._lv_check(vg_name, lv_snapshot_name) and \
self._lv_check(vg_name, lv_name):
self._lv_revert(vg_name, lv_snapshot_name)
self._lv_take_snapshot(vg_name, lv_name, lv_snapshot_name,
lv_snapshot_size)
elif self._lv_check(vg_name, lv_snapshot_name) and \
not self._lv_check(vg_name, lv_name):
raise error.TestError("Snapshot origin not found")
elif not self._lv_check(vg_name, lv_snapshot_name) and \
self._lv_check(vg_name, lv_name):
self._lv_take_snapshot(vg_name, lv_name, lv_snapshot_name,
lv_snapshot_size)
else:
self._lv_create(vg_name, lv_name, lv_size)
_______________________________________________
Autotest-kernel mailing list
[email protected]
https://www.redhat.com/mailman/listinfo/autotest-kernel
