| index : archinstall32 | |
| Archlinux32 installer | gitolite user |
| summaryrefslogtreecommitdiff |
| -rw-r--r-- | archinstall/lib/disk/filesystem.py | 2 | ||||
| -rw-r--r-- | archinstall/lib/disk/helpers.py | 88 | ||||
| -rw-r--r-- | archinstall/lib/disk/partition.py | 125 |
diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py index bdfa502a..1e722ce5 100644 --- a/archinstall/lib/disk/filesystem.py +++ b/archinstall/lib/disk/filesystem.py @@ -113,7 +113,7 @@ class Filesystem: format_options = partition.get('options',[]) + partition.get('filesystem',{}).get('format_options',[]) disk_encryption: DiskEncryption = storage['arguments'].get('disk_encryption') - if partition in disk_encryption.partitions: + if disk_encryption and partition in disk_encryption.partitions: if not partition['device_instance']: raise DiskError(f"Internal error caused us to loose the partition. Please report this issue upstream!") diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py index a5164b76..aea794f5 100644 --- a/archinstall/lib/disk/helpers.py +++ b/archinstall/lib/disk/helpers.py @@ -212,6 +212,47 @@ def all_disks() -> List[BlockDevice]: log(f"[Deprecated] archinstall.all_disks() is deprecated. Use archinstall.all_blockdevices() with the appropriate filters instead.", level=logging.WARNING, fg="yellow") return all_blockdevices(partitions=False, mappers=False) +def get_blockdevice_info(device_path, exclude_iso_dev :bool = True) -> Dict[str, Any]: + for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']): + partprobe(device_path) + time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * retry_attempt)) + + try: + if exclude_iso_dev: + # exclude all devices associated with the iso boot locations + iso_devs = ['/run/archiso/airootfs', '/run/archiso/bootmnt'] + + try: + lsblk_info = get_lsblk_info(device_path) + except DiskError: + continue + + if any([dev in lsblk_info.mountpoints for dev in iso_devs]): + continue + + information = blkid(f'blkid -p -o export {device_path}') + return enrich_blockdevice_information(information) + except SysCallError as ex: + if ex.exit_code in (512, 2): + # Assume that it's a loop device, and try to get info on it + try: + resolved_device_name = device_path.readlink().name + except OSError: + resolved_device_name = device_path.name + + try: + information = get_loop_info(device_path) + if not information: + raise SysCallError(f"Could not get loop information for {resolved_device_name}", exit_code=1) + return enrich_blockdevice_information(information) + + except SysCallError: + information = get_blockdevice_uevent(resolved_device_name) + return enrich_blockdevice_information(information) + else: + # We could not reliably get any information, perhaps the disk is clean of information? + if retry_attempt == storage['DISK_RETRY_ATTEMPTS'] - 1: + raise ex def all_blockdevices( mappers: bool = False, @@ -230,40 +271,18 @@ def all_blockdevices( # we'll iterate the /sys/class definitions and find the information # from there. for block_device in glob.glob("/sys/class/block/*"): - device_path = pathlib.Path(f"/dev/{pathlib.Path(block_device).readlink().name}") + try: + device_path = pathlib.Path(f"/dev/{pathlib.Path(block_device).readlink().name}") + except FileNotFoundError: + log(f"Unknown device found by '/sys/class/block/*', ignoring: {device_path}", level=logging.WARNING, fg="yellow") if device_path.exists() is False: log(f"Unknown device found by '/sys/class/block/*', ignoring: {device_path}", level=logging.WARNING, fg="yellow") continue - try: - if exclude_iso_dev: - # exclude all devices associated with the iso boot locations - iso_devs = ['/run/archiso/airootfs', '/run/archiso/bootmnt'] - lsblk_info = get_lsblk_info(device_path) - if any([dev in lsblk_info.mountpoints for dev in iso_devs]): - continue - - information = blkid(f'blkid -p -o export {device_path}') - except SysCallError as ex: - if ex.exit_code in (512, 2): - # Assume that it's a loop device, and try to get info on it - try: - information = get_loop_info(device_path) - if not information: - print("Exit code for blkid -p -o export was:", ex.exit_code) - raise SysCallError("Could not get loop information", exit_code=1) - - except SysCallError: - print("Not a loop device, trying uevent rules.") - information = get_blockdevice_uevent(pathlib.Path(block_device).readlink().name) - else: - # We could not reliably get any information, perhaps the disk is clean of information? - print("Raising ex because:", ex.exit_code) - raise ex - # return instances - - information = enrich_blockdevice_information(information) + information = get_blockdevice_info(device_path) + if not information: + continue for path, path_info in information.items(): if path_info.get('DMCRYPT_NAME'): @@ -409,7 +428,6 @@ def get_partitions_in_use(mountpoint :str) -> Dict[str, Any]: return {} output = json.loads(output) - # print(output) mounts = {} @@ -421,11 +439,13 @@ def get_partitions_in_use(mountpoint :str) -> Dict[str, Any]: continue if isinstance(blockdev, Partition): - for blockdev_mountpoint in blockdev.mountpoints: - block_devices_mountpoints[blockdev_mountpoint] = blockdev + if blockdev.mountpoints: + for blockdev_mountpoint in blockdev.mountpoints: + block_devices_mountpoints[blockdev_mountpoint] = blockdev else: - for blockdev_mountpoint in blockdev.mount_information: - block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev + if blockdev.mount_information: + for blockdev_mountpoint in blockdev.mount_information: + block_devices_mountpoints[blockdev_mountpoint['target']] = blockdev log(f'Filtering available mounts {block_devices_mountpoints} to those under {mountpoint}', level=logging.DEBUG) diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py index 9febf102..12b1a9a6 100644 --- a/archinstall/lib/disk/partition.py +++ b/archinstall/lib/disk/partition.py @@ -98,17 +98,18 @@ class Partition: if mountpoint: self.mount(mountpoint) - self._partition_info = self._fetch_information() - - if not autodetect_filesystem and filesystem: - self._partition_info.filesystem_type = filesystem + try: + self._partition_info = self._fetch_information() + + if not autodetect_filesystem and filesystem: + self._partition_info.filesystem_type = filesystem - if self._partition_info.filesystem_type == 'crypto_LUKS': - self._encrypted = True + if self._partition_info.filesystem_type == 'crypto_LUKS': + self._encrypted = True + except DiskError: + self._partition_info = None - # I hate doint this but I'm currently unsure where this - # is acutally used to be able to fix the typing issues properly - @typing.no_type_check + @typing.no_type_check # I hate doint this but I'm currently unsure where this is used. def __lt__(self, left_comparitor :BlockDevice) -> bool: if type(left_comparitor) == Partition: left_comparitor = left_comparitor.path @@ -120,14 +121,17 @@ class Partition: def __repr__(self, *args :str, **kwargs :str) -> str: mount_repr = '' - if mountpoint := self._partition_info.get_first_mountpoint(): - mount_repr = f", mounted={mountpoint}" - elif self._target_mountpoint: - mount_repr = f", rel_mountpoint={self._target_mountpoint}" + if self._partition_info: + if mountpoint := self._partition_info.get_first_mountpoint(): + mount_repr = f", mounted={mountpoint}" + elif self._target_mountpoint: + mount_repr = f", rel_mountpoint={self._target_mountpoint}" classname = self.__class__.__name__ - if self._encrypted: + if not self._partition_info: + return f'{classname}(path={self._path})' + elif self._encrypted: return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, parent={self.real_device}, fs={self._partition_info.filesystem_type}{mount_repr})' else: return f'{classname}(path={self._path}, size={self.size}, PARTUUID={self.part_uuid}, fs={self._partition_info.filesystem_type}{mount_repr})' @@ -146,7 +150,7 @@ class Partition: 'encrypted': self._encrypted, 'start': self.start, 'size': self.end, - 'filesystem': self._partition_info.filesystem_type + 'filesystem': self._partition_info.filesystem_type if self._partition_info else 'Unknown' } return partition_info @@ -164,34 +168,38 @@ class Partition: 'start': self.start, 'size': self.end, 'filesystem': { - 'format': self._partition_info.filesystem_type + 'format': self._partition_info.filesystem_type if self._partition_info else 'None' } } def _call_lsblk(self) -> Dict[str, Any]: - self.partprobe() - # This sleep might be overkill, but lsblk is known to - # work against a chaotic cache that can change during call - # causing no information to be returned (blkid is better) - # time.sleep(1) + for retry_attempt in range(storage['DISK_RETRY_ATTEMPTS']): + self.partprobe() + time.sleep(max(0.1, storage['DISK_TIMEOUTS'] * retry_attempt)) # TODO: Remove, we should be relying on blkid instead of lsblk + # This sleep might be overkill, but lsblk is known to + # work against a chaotic cache that can change during call + # causing no information to be returned (blkid is better) + # time.sleep(1) - # TODO: Maybe incorporate a re-try system here based on time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1))) + # TODO: Maybe incorporate a re-try system here based on time.sleep(max(0.1, storage.get('DISK_TIMEOUTS', 1))) - try: - output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8') - except SysCallError as error: - # It appears as if lsblk can return exit codes like 8192 to indicate something. - # But it does return output so we'll try to catch it. - output = error.worker.decode('UTF-8') - - if output: try: - lsblk_info = json.loads(output) - return lsblk_info - except json.decoder.JSONDecodeError: - log(f"Could not decode JSON: {output}", fg="red", level=logging.ERROR) - - raise DiskError(f'Failed to read disk "{self.device_path}" with lsblk') + output = SysCommand(f"lsblk --json -b -o+LOG-SEC,SIZE,PTTYPE,PARTUUID,UUID,FSTYPE {self.device_path}").decode('UTF-8') + except SysCallError as error: + # It appears as if lsblk can return exit codes like 8192 to indicate something. + # But it does return output in stderr so we'll try to catch it minus the message/info. + output = error.worker.decode('UTF-8') + if '{' in output: + output = output[output.find('{'):] + + if output: + try: + lsblk_info = json.loads(output) + return lsblk_info + except json.decoder.JSONDecodeError: + log(f"Could not decode JSON: {output}", fg="red", level=logging.ERROR) + + raise DiskError(f'Failed to get partition information "{self.device_path}" with lsblk') def _call_sfdisk(self) -> Dict[str, Any]: output = SysCommand(f"sfdisk --json {self.block_device.path}").decode('UTF-8') @@ -212,9 +220,12 @@ class Partition: lsblk_info = self._call_lsblk() sfdisk_info = self._call_sfdisk() - if not (device := lsblk_info.get('blockdevices', [None])[0]): + if not (device := lsblk_info.get('blockdevices', [])): raise DiskError(f'Failed to retrieve information for "{self.device_path}" with lsblk') + # Grab the first (and only) block device in the list as we're targeting a specific partition + device = device[0] + mountpoints = [Path(mountpoint) for mountpoint in device['mountpoints'] if mountpoint] bootable = sfdisk_info.get('bootable', False) or sfdisk_info.get('type', '') == 'C12A7328-F81F-11D2-BA4B-00A0C93EC93B' @@ -243,7 +254,8 @@ class Partition: @property def filesystem(self) -> str: - return self._partition_info.filesystem_type + if self._partition_info: + return self._partition_info.filesystem_type @property def mountpoint(self) -> Optional[Path]: @@ -253,43 +265,51 @@ class Partition: @property def mountpoints(self) -> List[Path]: - return self._partition_info.mountpoints + if self._partition_info: + return self._partition_info.mountpoints @property def sector_size(self) -> int: - return self._partition_info.sector_size + if self._partition_info: + return self._partition_info.sector_size @property def start(self) -> Optional[int]: - return self._partition_info.start + if self._partition_info: + return self._partition_info.start @property def end(self) -> Optional[int]: - return self._partition_info.end + if self._partition_info: + return self._partition_info.end @property def end_sectors(self) -> Optional[int]: - start = self._partition_info.start - end = self._partition_info.end - if start and end: - return start + end - return None + if self._partition_info: + start = self._partition_info.start + end = self._partition_info.end + if start and end: + return start + end @property def size(self) -> Optional[float]: - return self._partition_info.size + if self._partition_info: + return self._partition_info.size @property def boot(self) -> bool: - return self._partition_info.bootable + if self._partition_info: + return self._partition_info.bootable @property def partition_type(self) -> Optional[str]: - return self._partition_info.pttype + if self._partition_info: + return self._partition_info.pttype @property def part_uuid(self) -> str: - return self._partition_info.partuuid + if self._partition_info: + return self._partition_info.partuuid @property def uuid(self) -> Optional[str]: @@ -355,7 +375,8 @@ class Partition: log(f"Could not get PARTUUID of partition using 'blkid -s PARTUUID -o value {self.device_path}': {error}") - return self._partition_info.uuid + if self._partition_info: + return self._partition_info.uuid @property def encrypted(self) -> Union[bool, None]: |