LVM on Luks

archinstall/lib/disk/__init__.py

@@ -35,7 +35,8 @@
 	CleanType,
 	get_lsblk_info,
 	get_all_lsblk_info,
-	get_lsblk_by_mountpoint
+	get_lsblk_by_mountpoint,
+	get_luks_mapper_lsblk_info
 )
 from .encryption_menu import (
 	select_encryption_type,

archinstall/lib/disk/device_handler.py

@@ -19,7 +19,7 @@
 	FilesystemType, Unit, PartitionTable,
 	ModificationStatus, get_lsblk_info, LsblkInfo,
 	_BtrfsSubvolumeInfo, get_all_lsblk_info, DiskEncryption, LvmVolumeGroup, LvmVolume, Size, LvmGroupInfo,
-	SectorSize, LvmVolumeInfo
+	SectorSize, LvmVolumeInfo, LvmPVInfo
 )
 
 from ..exceptions import DiskError, UnknownFilesystemFormat
@@ -138,7 +138,7 @@ def get_unique_path_for_device(self, dev_path: Path) -> Optional[Path]:
 		paths = Path('/dev/disk/by-id').glob('*')
 		linked_targets = {p.resolve(): p for p in paths}
 		linked_wwn_targets = {p: linked_targets[p] for p in linked_targets
-			if p.name.startswith('wwn-') or p.name.startswith('nvme-eui.')}
+							  if p.name.startswith('wwn-') or p.name.startswith('nvme-eui.')}
 
 		if dev_path in linked_wwn_targets:
 			return linked_wwn_targets[dev_path]
@@ -297,54 +297,70 @@ def format_encrypted(
 		info(f'luks2 locking device: {dev_path}')
 		luks_handler.lock()
 
-	def _lvm_info(self, cmd: str, name: str, info_type: Literal['lv', 'vg']) -> Optional[Any]:
+	def _lvm_info(
+		self,
+		cmd: str,
+		info_type: Literal['lv', 'vg', 'pvseg']
+	) -> Optional[Any]:
 		raw_info = SysCommand(cmd).decode().split('\n')
 
 		# for whatever reason the output sometimes contains
 		# "File descriptor X leaked leaked on vgs invocation
 		data = '\n'.join([raw for raw in raw_info if 'File descriptor' not in raw])
 
-		debug(f'LVM raw: {data}')
+		debug(f'LVM info: {data}')
 
 		reports = json.loads(data)
 
-		type_name = f'{info_type}_name'
-
 		for report in reports['report']:
-			entries = report.get(info_type, [])
-			for entry in entries:
-				if entry[type_name] == name:
-					lv_size = int(entry[f'{info_type}_size'][:-1])
-					size = Size(lv_size, Unit.B, SectorSize.default())
-
-					match info_type:
-						case 'lv':
-							return LvmVolumeInfo(
-								lv_name=entry['lv_name'],
-								vg_name=entry['vg_name'],
-								lv_size=size
-							)
-						case 'vg':
-							return LvmGroupInfo(
-								vg_size=size,
-								vg_uuid=entry['vg_uuid']
-							)
+			if len(report) != 1:
+				raise ValueError(f'Report does not contain single entry')
+
+			entry = report[info_type][0]
+
+			match info_type:
+				case 'pvseg':
+					return LvmPVInfo(
+						pv_name=entry['pv_name'],
+						lv_name=entry['lv_name'],
+						vg_name=entry['vg_name'],
+					)
+				case 'lv':
+					return LvmVolumeInfo(
+						lv_name=entry['lv_name'],
+						vg_name=entry['vg_name'],
+						lv_size=Size(int(entry[f'lv_size'][:-1]), Unit.B, SectorSize.default())
+					)
+				case 'vg':
+					return LvmGroupInfo(
+						vg_uuid=entry['vg_uuid'],
+						vg_size=Size(int(entry[f'vg_size'][:-1]), Unit.B, SectorSize.default())
+					)
+
 		return None
 
 	def lvm_vol_info(self, lv_name: str) -> Optional[LvmVolumeInfo]:
 		cmd = 'lvs --reportformat json ' \
-			'--unit B ' \
-			f'-S lv_name={lv_name}'
+			  '--unit B ' \
+			  f'-S lv_name={lv_name}'
 
-		return self._lvm_info(cmd, lv_name, 'lv')
+		return self._lvm_info(cmd, 'lv')
 
 	def lvm_group_info(self, vg_name: str) -> Optional[LvmGroupInfo]:
 		cmd = 'vgs --reportformat json ' \
-			'--unit B ' \
-			'-o vg_name,vg_uuid,vg_size ' \
-			f'-S vg_name={vg_name}'
+			  '--unit B ' \
+			  '-o vg_name,vg_uuid,vg_size ' \
+			  f'-S vg_name={vg_name}'
+
+		return self._lvm_info(cmd, 'vg')
+
+	def lvm_pvseg_info(self, vg_name: str, lv_name: str) -> Optional[LvmPVInfo]:
+		cmd = 'pvs ' \
+			  '--segments -o+lv_name,vg_name ' \
+			  f'-S vg_name={vg_name},lv_name={lv_name} ' \
+			  '--reportformat json '
 
-		return self._lvm_info(cmd, vg_name, 'vg')
+		return self._lvm_info(cmd, 'pvseg')
 
 	def lvm_vol_change(self, vol: LvmVolume, activate: bool):
 		active_flag = 'y' if activate else 'n'
@@ -405,6 +421,7 @@ def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Optional[Size]
 		worker.poll()
 		worker.write(b'y\n', line_ending=False)
 
+		volume.vg_name = vg_name
 		volume.dev_path = Path(f'/dev/{vg_name}/{volume.name}')
 
 	def _perform_partitioning(

archinstall/lib/disk/device_model.py

@@ -886,6 +886,7 @@ def table_data(self) -> Dict[str, Any]:
 
 class LvmLayoutType(Enum):
 	Default = 'default'
+
 	# Manual = 'manual_lvm'
 
 	def display_msg(self) -> str:
@@ -925,6 +926,9 @@ def parse_arg(arg: Dict[str, Any], disk_config: DiskLayoutConfiguration) -> LvmV
 			[LvmVolume.parse_arg(vol) for vol in arg['volumes']]
 		)
 
+	def contains_lv(self, lv: LvmVolume) -> bool:
+		return lv in self.volumes
+
 
 class LvmVolumeStatus(Enum):
 	Exist = 'existing'
@@ -943,14 +947,16 @@ class LvmVolume:
 	mount_options: List[str] = field(default_factory=list)
 	btrfs_subvols: List[SubvolumeModification] = field(default_factory=list)
 
+	# volume group name
+	vg_name: Optional[str] = None
+	# mapper device path /dev/<vg>/<vol>
+	dev_path: Optional[Path] = None
+
 	def __post_init__(self):
 		# needed to use the object as a dictionary key due to hash func
 		if not hasattr(self, '_obj_id'):
 			self._obj_id = uuid.uuid4()
 
-	# mapper device path /dev/<vg>/<vol>
-	dev_path: Optional[Path] = None
-
 	def __hash__(self):
 		return hash(self._obj_id)
 
@@ -1062,6 +1068,13 @@ class LvmVolumeInfo:
 	lv_size: Size
 
 
+@dataclass
+class LvmPVInfo:
+	pv_name: str
+	lv_name: str
+	vg_name: str
+
+
 @dataclass
 class LvmConfiguration:
 	config_type: LvmLayoutType
@@ -1118,6 +1131,16 @@ def get_root_volume(self) -> Optional[LvmVolume]:
 
 		return None
 
+
+# def get_lv_crypt_uuid(self, lv: LvmVolume, encryption: EncryptionType) -> str:
+# 	"""
+# 	Find the LUKS superblock UUID for the device that
+# 	contains the given logical volume
+# 	"""
+# 	for vg in self.vol_groups:
+# 		if vg.contains_lv(lv):
+
+
 @dataclass
 class DeviceModification:
 	device: BDevice
@@ -1409,16 +1432,25 @@ def _clean_field(name: str, clean_type: CleanType) -> str:
 			return name.replace('_percentage', '%').replace('_', '-')
 
 
-def _fetch_lsblk_info(dev_path: Optional[Union[Path, str]] = None, retry: int = 3) -> List[LsblkInfo]:
+def _fetch_lsblk_info(
+	dev_path: Optional[Union[Path, str]] = None,
+	reverse: bool = False,
+	full_dev_path: bool = False,
+	retry: int = 3
+) -> List[LsblkInfo]:
 	fields = [_clean_field(f, CleanType.Lsblk) for f in LsblkInfo.fields()]
 	lsblk_fields = ','.join(fields)
 
 	if not dev_path:
 		dev_path = ''
 
+	cmd = f'lsblk --json -b -o+{lsblk_fields} {dev_path}'
+	cmd += ' -s' if reverse else ''
+	cmd += ' -p ' if full_dev_path else ''
+
 	for retry_attempt in range(retry + 1):
 		try:
-			result = SysCommand(f'lsblk --json -b -o+{lsblk_fields} {dev_path}').decode()
+			result = SysCommand(cmd).decode()
 			break
 		except SysCallError as err:
 			# Get the output minus the message/info from lsblk if it returns a non-zero exit code.
@@ -1474,3 +1506,10 @@ def _check(infos: List[LsblkInfo]) -> List[LsblkInfo]:
 
 	all_info = get_all_lsblk_info()
 	return _check(all_info)
+
+
+def get_luks_mapper_lsblk_info(mapper_dev_path: Union[Path, str]) -> List[LsblkInfo]:
+	"""
+	Find the LUKS superblock UUID for the given mapper path
+	"""
+	return _fetch_lsblk_info(mapper_dev_path, reverse=True, full_dev_path=True)