osbuild-image-info: extract discovery of LV names into a function

Extract the code that discovers LVM LV names for a given VG, from
`discover_lvm()` into a separate function `lvm_lvs_for_vg()`. This
improves the readability of the code. In addition, some values returned
by the `lvdisplay` invocation were never used. Don't request them and
simplify the code. Rename variables that hold LV names to clearly
express that.

Signed-off-by: Tomáš Hozza <thozza@redhat.com>
This commit is contained in:
Tomáš Hozza 2025-01-27 10:44:58 +01:00 committed by Tomáš Hozza
parent 9bdde1bc80
commit 0fba11369c

View file

@ -19,7 +19,7 @@ import tempfile
import time import time
import xml.etree.ElementTree import xml.etree.ElementTree
from collections import OrderedDict from collections import OrderedDict
from typing import Any, Dict from typing import Any, Dict, List
import jsonschema import jsonschema
import yaml import yaml
@ -2486,26 +2486,13 @@ def volume_group_for_device(device: str) -> str:
return vg_name return vg_name
def ensure_device_file(path: str, major: int, minor: int): def lvm_lvs_for_vg(vg_name: str) -> List[str]:
"""Ensure the device file with the given major, minor exists""" """
os.makedirs(os.path.dirname(path), exist_ok=True) Get the list of logical volumes for a given volume group.
if not os.path.exists(path): """
os.mknod(path, 0o600 | stat.S_IFBLK, os.makedev(major, minor))
def discover_lvm(dev: str, parent: devices.Device, devmgr: devices.DeviceManager):
# find the volume group name for the device file
vg_name = volume_group_for_device(dev)
# activating LVM is done OSBuild side.
# However we still have to get OSBuild the name of the VG to open
# Find all logical volumes in the volume group
cmd = [ cmd = [
"lvdisplay", "-C", "--noheadings", "lvdisplay", "-C", "--noheadings",
"-o", "lv_name,path,lv_kernel_major,lv_kernel_minor", "-o", "lv_name", vg_name
"--separator", ";",
vg_name
] ]
res = subprocess.run(cmd, res = subprocess.run(cmd,
@ -2517,27 +2504,40 @@ def discover_lvm(dev: str, parent: devices.Device, devmgr: devices.DeviceManager
if res.returncode != 0: if res.returncode != 0:
raise RuntimeError(res.stderr.strip()) raise RuntimeError(res.stderr.strip())
data = res.stdout.strip() return [lv.strip() for lv in res.stdout.split("\n")]
parsed = list(map(lambda line: line.split(";"), data.split("\n")))
volumes = OrderedDict()
def ensure_device_file(path: str, major: int, minor: int):
"""Ensure the device file with the given major, minor exists"""
os.makedirs(os.path.dirname(path), exist_ok=True)
if not os.path.exists(path):
os.mknod(path, 0o600 | stat.S_IFBLK, os.makedev(major, minor))
def discover_lvm(dev: str, parent: devices.Device, devmgr: devices.DeviceManager):
# find the volume group name for the device file
vg_name = volume_group_for_device(dev)
# NB: activating LVM is done by the OSBuild device implementation,
# however, the LV name must be passed to the OSBuild device implementation.
lv_names = lvm_lvs_for_vg(vg_name)
# NB: the order of the volumes is important, we want to mount the root
# volume first, so that we can mount the other volumes on top of it.
volumes = OrderedDict()
# devices_map stores for each device path onto the system the corresponding # devices_map stores for each device path onto the system the corresponding
# OSBuild's Device object # OSBuild's Device object
devices_map = {} devices_map = {}
for vol in parsed: for lv_name in lv_names:
vol = list(map(lambda v: v.strip(), vol))
assert len(vol) == 4
name, _, _, _ = vol
options = { options = {
"volume": name, "volume": lv_name,
} }
# Create an OSBuild device object for the LVM partition # Create an OSBuild device object for the LVM partition
info = index.get_module_info("Device", "org.osbuild.lvm2.lv") info = index.get_module_info("Device", "org.osbuild.lvm2.lv")
device = devices.Device( device = devices.Device(
name, lv_name,
info, info,
parent, parent,
options) options)
@ -2555,9 +2555,9 @@ def discover_lvm(dev: str, parent: devices.Device, devmgr: devices.DeviceManager
ensure_device_file(voldev, int(major), int(minor)) ensure_device_file(voldev, int(major), int(minor))
read_partition(voldev, info) read_partition(voldev, info)
volumes[name] = info volumes[lv_name] = info
if name.startswith("root"): if lv_name.startswith("root"):
volumes.move_to_end(name, last=False) volumes.move_to_end(lv_name, last=False)
# associate the device path with the Device object, we will need it to # associate the device path with the Device object, we will need it to
# mount later on. # mount later on.