| index : archinstall32 | |
| Archlinux32 installer | gitolite user |
| summaryrefslogtreecommitdiff |
| -rw-r--r-- | archinstall/lib/disk/btrfs/__init__.py | 56 | ||||
| -rw-r--r-- | archinstall/lib/disk/btrfs/btrfs_helpers.py | 136 | ||||
| -rw-r--r-- | archinstall/lib/disk/btrfs/btrfspartition.py | 109 | ||||
| -rw-r--r-- | archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py | 192 |
diff --git a/archinstall/lib/disk/btrfs/__init__.py b/archinstall/lib/disk/btrfs/__init__.py deleted file mode 100644 index a26e0160..00000000 --- a/archinstall/lib/disk/btrfs/__init__.py +++ /dev/null @@ -1,56 +0,0 @@ -from __future__ import annotations -import pathlib -import glob -import logging -from typing import Union, Dict, TYPE_CHECKING - -# https://stackoverflow.com/a/39757388/929999 -if TYPE_CHECKING: - from ...installer import Installer - -from .btrfs_helpers import ( - subvolume_info_from_path as subvolume_info_from_path, - find_parent_subvolume as find_parent_subvolume, - setup_subvolumes as setup_subvolumes, - mount_subvolume as mount_subvolume -) -from .btrfssubvolumeinfo import BtrfsSubvolumeInfo as BtrfsSubvolume -from .btrfspartition import BTRFSPartition as BTRFSPartition - -from ...exceptions import DiskError, Deprecated -from ...general import SysCommand -from ...output import log - - -def create_subvolume(installation: Installer, subvolume_location :Union[pathlib.Path, str]) -> bool: - """ - This function uses btrfs to create a subvolume. - - @installation: archinstall.Installer instance - @subvolume_location: a localized string or path inside the installation / or /boot for instance without specifying /mnt/boot - """ - - installation_mountpoint = installation.target - if type(installation_mountpoint) == str: - installation_mountpoint = pathlib.Path(installation_mountpoint) - # Set up the required physical structure - if type(subvolume_location) == str: - subvolume_location = pathlib.Path(subvolume_location) - - target = installation_mountpoint / subvolume_location.relative_to(subvolume_location.anchor) - - # Difference from mount_subvolume: - # We only check if the parent exists, since we'll run in to "target path already exists" otherwise - if not target.parent.exists(): - target.parent.mkdir(parents=True) - - if glob.glob(str(target / '*')): - raise DiskError(f"Cannot create subvolume at {target} because it contains data (non-empty folder target)") - - # Remove the target if it exists - if target.exists(): - target.rmdir() - - log(f"Creating a subvolume on {target}", level=logging.INFO) - if (cmd := SysCommand(f"btrfs subvolume create {target}")).exit_code != 0: - raise DiskError(f"Could not create a subvolume at {target}: {cmd}") diff --git a/archinstall/lib/disk/btrfs/btrfs_helpers.py b/archinstall/lib/disk/btrfs/btrfs_helpers.py deleted file mode 100644 index f6d2734a..00000000 --- a/archinstall/lib/disk/btrfs/btrfs_helpers.py +++ /dev/null @@ -1,136 +0,0 @@ -import logging -import re -from pathlib import Path -from typing import Optional, Dict, Any, TYPE_CHECKING - -from ...models.subvolume import Subvolume -from ...exceptions import SysCallError, DiskError -from ...general import SysCommand -from ...output import log -from ...plugins import plugins -from ..helpers import get_mount_info -from .btrfssubvolumeinfo import BtrfsSubvolumeInfo - -if TYPE_CHECKING: - from .btrfspartition import BTRFSPartition - from ...installer import Installer - - -class fstab_btrfs_compression_plugin(): - def __init__(self, partition_dict): - self.partition_dict = partition_dict - - def on_genfstab(self, installation): - with open(f"{installation.target}/etc/fstab", 'r') as fh: - fstab = fh.read() - - # Replace the {installation}/etc/fstab with entries - # using the compress=zstd where the mountpoint has compression set. - with open(f"{installation.target}/etc/fstab", 'w') as fh: - for line in fstab.split('\n'): - # So first we grab the mount options by using subvol=.*? as a locator. - # And we also grab the mountpoint for the entry, for instance /var/log - if (subvoldef := re.findall(',.*?subvol=.*?[\t ]', line)) and (mountpoint := re.findall('[\t ]/.*?[\t ]', line)): - for subvolume in self.partition_dict.get('btrfs', {}).get('subvolumes', []): - # We then locate the correct subvolume and check if it's compressed - if subvolume.compress and subvolume.mountpoint == mountpoint[0].strip(): - # We then sneak in the compress=zstd option if it doesn't already exist: - # We skip entries where compression is already defined - if ',compress=zstd,' not in line: - line = line.replace(subvoldef[0], f",compress=zstd{subvoldef[0]}") - break - - fh.write(f"{line}\n") - - return True - - -def mount_subvolume(installation: 'Installer', device: 'BTRFSPartition', subvolume: Subvolume): - # we normalize the subvolume name (getting rid of slash at the start if exists. - # In our implementation has no semantic load. - # Every subvolume is created from the top of the hierarchy- and simplifies its further use - name = subvolume.name.lstrip('/') - mountpoint = Path(subvolume.mountpoint) - installation_target = Path(installation.target) - - mountpoint = installation_target / mountpoint.relative_to(mountpoint.anchor) - mountpoint.mkdir(parents=True, exist_ok=True) - mount_options = subvolume.options + [f'subvol={name}'] - - log(f"Mounting subvolume {name} on {device} to {mountpoint}", level=logging.INFO, fg="gray") - SysCommand(f"mount {device.path} {mountpoint} -o {','.join(mount_options)}") - - -def setup_subvolumes(installation: 'Installer', partition_dict: Dict[str, Any]): - log(f"Setting up subvolumes: {partition_dict['btrfs']['subvolumes']}", level=logging.INFO, fg="gray") - - for subvolume in partition_dict['btrfs']['subvolumes']: - # we normalize the subvolume name (getting rid of slash at the start if exists. In our implementation has no semantic load. - # Every subvolume is created from the top of the hierarchy- and simplifies its further use - name = subvolume.name.lstrip('/') - - # We create the subvolume using the BTRFSPartition instance. - # That way we ensure not only easy access, but also accurate mount locations etc. - partition_dict['device_instance'].create_subvolume(name, installation=installation) - - # Make the nodatacow processing now - # It will be the main cause of creation of subvolumes which are not to be mounted - # it is not an options which can be established by subvolume (but for whole file systems), and can be - # set up via a simple attribute change in a directory (if empty). And here the directories are brand new - if subvolume.nodatacow: - if (cmd := SysCommand(f"chattr +C {installation.target}/{name}")).exit_code != 0: - raise DiskError(f"Could not set nodatacow attribute at {installation.target}/{name}: {cmd}") - - # Make the compress processing now - # it is not an options which can be established by subvolume (but for whole file systems), and can be - # set up via a simple attribute change in a directory (if empty). And here the directories are brand new - # in this way only zstd compression is activaded - # TODO WARNING it is not clear if it should be a standard feature, so it might need to be deactivated - - if subvolume.compress: - if not any(['compress' in filesystem_option for filesystem_option in partition_dict.get('filesystem', {}).get('mount_options', [])]): - if (cmd := SysCommand(f"chattr +c {installation.target}/{name}")).exit_code != 0: - raise DiskError(f"Could not set compress attribute at {installation.target}/{name}: {cmd}") - - if 'fstab_btrfs_compression_plugin' not in plugins: - plugins['fstab_btrfs_compression_plugin'] = fstab_btrfs_compression_plugin(partition_dict) - - -def subvolume_info_from_path(path: Path) -> Optional[BtrfsSubvolumeInfo]: - try: - subvolume_name = '' - result = {} - for index, line in enumerate(SysCommand(f"btrfs subvolume show {path}")): - if index == 0: - subvolume_name = line.strip().decode('UTF-8') - continue - - if b':' in line: - key, value = line.strip().decode('UTF-8').split(':', 1) - - # A bit of a hack, until I figure out how @dataclass - # allows for hooking in a pre-processor to do this we have to do it here: - result[key.lower().replace(' ', '_').replace('(s)', 's')] = value.strip() - - return BtrfsSubvolumeInfo(**{'full_path' : path, 'name' : subvolume_name, **result}) # type: ignore - except SysCallError as error: - log(f"Could not retrieve subvolume information from {path}: {error}", level=logging.WARNING, fg="orange") - - return None - - -def find_parent_subvolume(path: Path, filters=[]) -> Optional[BtrfsSubvolumeInfo]: - # A root path cannot have a parent - if str(path) == '/': - return None - - if found_mount := get_mount_info(str(path.parent), traverse=True, ignore=filters): - if not (subvolume := subvolume_info_from_path(found_mount['target'])): - if found_mount['target'] == '/': - return None - - return find_parent_subvolume(path.parent, filters=[*filters, found_mount['target']]) - - return subvolume - - return None diff --git a/archinstall/lib/disk/btrfs/btrfspartition.py b/archinstall/lib/disk/btrfs/btrfspartition.py deleted file mode 100644 index d04c9b98..00000000 --- a/archinstall/lib/disk/btrfs/btrfspartition.py +++ /dev/null @@ -1,109 +0,0 @@ -import glob -import pathlib -import logging -from typing import Optional, TYPE_CHECKING - -from ...exceptions import DiskError -from ...storage import storage -from ...output import log -from ...general import SysCommand -from ..partition import Partition -from ..helpers import findmnt -from .btrfs_helpers import ( - subvolume_info_from_path -) - -if TYPE_CHECKING: - from ...installer import Installer - from .btrfssubvolumeinfo import BtrfsSubvolumeInfo - - -class BTRFSPartition(Partition): - def __init__(self, *args, **kwargs): - Partition.__init__(self, *args, **kwargs) - - @property - def subvolumes(self): - for filesystem in findmnt(pathlib.Path(self.path), recurse=True).get('filesystems', []): - if '[' in filesystem.get('source', ''): - yield subvolume_info_from_path(filesystem['target']) - - def iterate_children(struct): - for c in struct.get('children', []): - if '[' in child.get('source', ''): - yield subvolume_info_from_path(c['target']) - - for sub_child in iterate_children(c): - yield sub_child - - for child in iterate_children(filesystem): - yield child - - def create_subvolume(self, subvolume :pathlib.Path, installation :Optional['Installer'] = None) -> 'BtrfsSubvolumeInfo': - """ - Subvolumes have to be created within a mountpoint. - This means we need to get the current installation target. - After we get it, we need to verify it is a btrfs subvolume filesystem. - Finally, the destination must be empty. - """ - - # Allow users to override the installation session - if not installation: - installation = storage.get('installation_session') - - # Determain if the path given, is an absolute path or a relative path. - # We do this by checking if the path contains a known mountpoint. - if str(subvolume)[0] == '/': - if filesystems := findmnt(subvolume, traverse=True).get('filesystems'): - if (target := filesystems[0].get('target')) and target != '/' and str(subvolume).startswith(target): - # Path starts with a known mountpoint which isn't / - # Which means it's an absolute path to a mounted location. - pass - else: - # Since it's not an absolute position with a known start. - # We omit the anchor ('/' basically) and make sure it's appendable - # to the installation.target later - subvolume = subvolume.relative_to(subvolume.anchor) - # else: We don't need to do anything about relative paths, they should be appendable to installation.target as-is. - - # If the subvolume is not absolute, then we do two checks: - # 1. Check if the partition itself is mounted somewhere, and use that as a root - # 2. Use an active Installer().target as the root, assuming it's filesystem is btrfs - # If both above fail, we need to warn the user that such setup is not supported. - if str(subvolume)[0] != '/': - if self.mountpoint is None and installation is None: - raise DiskError("When creating a subvolume on BTRFSPartition()'s, you need to either initiate a archinstall.Installer() or give absolute paths when creating the subvoulme.") - elif self.mountpoint: - subvolume = self.mountpoint / subvolume - elif installation: - ongoing_installation_destination = installation.target - if type(ongoing_installation_destination) == str: - ongoing_installation_destination = pathlib.Path(ongoing_installation_destination) - - subvolume = ongoing_installation_destination / subvolume - - subvolume.parent.mkdir(parents=True, exist_ok=True) - - # <!-- - # We perform one more check from the given absolute position. - # And we traverse backwards in order to locate any if possible subvolumes above - # our new btrfs subvolume. This is because it needs to be mounted under it to properly - # function. - # if btrfs_parent := find_parent_subvolume(subvolume): - # print('Found parent:', btrfs_parent) - # --> - - log(f'Attempting to create subvolume at {subvolume}', level=logging.DEBUG, fg="grey") - - if glob.glob(str(subvolume / '*')): - raise DiskError(f"Cannot create subvolume at {subvolume} because it contains data (non-empty folder target is not supported by BTRFS)") - # Ideally we would like to check if the destination is already a subvolume. - # But then we would need the mount-point at this stage as well. - # So we'll comment out this check: - # elif subvolinfo := subvolume_info_from_path(subvolume): - # raise DiskError(f"Destination {subvolume} is already a subvolume: {subvolinfo}") - - # And deal with it here: - SysCommand(f"btrfs subvolume create {subvolume}") - - return subvolume_info_from_path(subvolume) diff --git a/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py b/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py deleted file mode 100644 index 5f5bdea6..00000000 --- a/archinstall/lib/disk/btrfs/btrfssubvolumeinfo.py +++ /dev/null @@ -1,192 +0,0 @@ -import pathlib -import datetime -import logging -import string -import random -import shutil -from dataclasses import dataclass -from typing import Optional, List# , TYPE_CHECKING -from functools import cached_property - -# if TYPE_CHECKING: -# from ..blockdevice import BlockDevice - -from ...exceptions import DiskError -from ...general import SysCommand -from ...output import log -from ...storage import storage - - -@dataclass -class BtrfsSubvolumeInfo: - full_path :pathlib.Path - name :str - uuid :str - parent_uuid :str - creation_time :datetime.datetime - subvolume_id :int - generation :int - gen_at_creation :int - parent_id :int - top_level_id :int - send_transid :int - send_time :datetime.datetime - receive_transid :int - received_uuid :Optional[str] = None - flags :Optional[str] = None - receive_time :Optional[datetime.datetime] = None - snapshots :Optional[List] = None - - def __post_init__(self): - self.full_path = pathlib.Path(self.full_path) - - # Convert "-" entries to `None` - if self.parent_uuid == "-": - self.parent_uuid = None - if self.received_uuid == "-": - self.received_uuid = None - if self.flags == "-": - self.flags = None - if self.receive_time == "-": - self.receive_time = None - if self.snapshots == "": - self.snapshots = [] - - # Convert timestamps into datetime workable objects (and preserve timezone by using ISO formats) - self.creation_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.creation_time)) - self.send_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.send_time)) - if self.receive_time: - self.receive_time = datetime.datetime.fromisoformat(self.convert_to_ISO_format(self.receive_time)) - - @property - def parent_subvolume(self): - from .btrfs_helpers import find_parent_subvolume - - return find_parent_subvolume(self.full_path) - - @property - def root(self) -> bool: - from .btrfs_helpers import subvolume_info_from_path - - # TODO: Make this function traverse storage['MOUNT_POINT'] and find the first - # occurrence of a mountpoint that is a btrfs volume instead of lazy assume / is a subvolume. - # It would also be nice if it could use findmnt(self.full_path) and traverse backwards - # finding the last occurrence of a subvolume which 'self' belongs to. - if volume := subvolume_info_from_path(storage['MOUNT_POINT']): - return self.full_path == volume.full_path - - return False - - @cached_property - def partition(self): - from ..helpers import findmnt, get_parent_of_partition, all_blockdevices - from ..partition import Partition - from ..blockdevice import BlockDevice - from ..mapperdev import MapperDev - from .btrfspartition import BTRFSPartition - from .btrfs_helpers import subvolume_info_from_path - - try: - # If the subvolume is mounted, it's pretty trivial to lookup the partition (parent) device. - if filesystem := findmnt(self.full_path).get('filesystems', []): - if source := filesystem[0].get('source', None): - # Strip away subvolume definitions from findmnt - if '[' in source: - source = source[:source.find('[')] - - if filesystem[0].get('fstype', '') == 'btrfs': - return BTRFSPartition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) - elif filesystem[0].get('source', '').startswith('/dev/mapper'): - return MapperDev(source) - else: - return Partition(source, BlockDevice(get_parent_of_partition(pathlib.Path(source)))) - except DiskError: - # Subvolume has never been mounted, we have no reliable way of finding where it is. - # But we have the UUID of the partition, and can begin looking for it by mounting - # all blockdevices that we can reliably support.. This is taxing tho and won't cover all devices. - - log(f"Looking up {self}, this might take time.", fg="orange", level=logging.WARNING) - for blockdevice, instance in all_blockdevices(mappers=True, partitions=True, error=True).items(): - if type(instance) in (Partition, MapperDev): - we_mounted_it = False - detection_mountpoint = instance.mountpoint - if not detection_mountpoint: - if type(instance) == Partition and instance.encrypted: - # TODO: Perhaps support unlocking encrypted volumes? - # This will cause a lot of potential user interactions tho. - log(f"Ignoring {blockdevice} because it's encrypted.", fg="gray", level=logging.DEBUG) - continue - - detection_mountpoint = pathlib.Path(f"/tmp/{''.join([random.choice(string.ascii_letters) for x in range(20)])}") - detection_mountpoint.mkdir(parents=True, exist_ok=True) - - instance.mount(str(detection_mountpoint)) - we_mounted_it = True - - if (filesystem := findmnt(detection_mountpoint)) and (filesystem := filesystem.get('filesystems', [])): - if subvolume := subvolume_info_from_path(filesystem[0]['target']): - if subvolume.uuid == self.uuid: - # The top level subvolume matched of ourselves, - # which means the instance we're iterating has the subvol we're looking for. - log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") - return instance - - def iterate_children(struct): - for child in struct.get('children', []): - if '[' in child.get('source', ''): - yield subvolume_info_from_path(child['target']) - - for sub_child in iterate_children(child): - yield sub_child - - for child in iterate_children(filesystem[0]): - if child.uuid == self.uuid: - # We found a child within the instance that has the subvol we're looking for. - log(f"Found the subvolume on device {instance}", level=logging.DEBUG, fg="gray") - return instance - - if we_mounted_it: - instance.unmount() - shutil.rmtree(detection_mountpoint) - - @cached_property - def mount_options(self) -> Optional[List[str]]: - from ..helpers import findmnt - - if filesystem := findmnt(self.full_path).get('filesystems', []): - return filesystem[0].get('options').split(',') - - def convert_to_ISO_format(self, time_string): - time_string_almost_done = time_string.replace(' ', 'T', 1).replace(' ', '') - iso_string = f"{time_string_almost_done[:-2]}:{time_string_almost_done[-2:]}" - return iso_string - - def mount(self, mountpoint :pathlib.Path, options=None, include_previously_known_options=True): - from ..helpers import findmnt - - try: - if mnt_info := findmnt(pathlib.Path(mountpoint), traverse=False): - log(f"Unmounting {mountpoint} as it was already mounted using {mnt_info}") - SysCommand(f"umount {mountpoint}") - except DiskError: - # No previously mounted device at the mountpoint - pass - - if not options: - options = [] - - try: - if include_previously_known_options and (cached_options := self.mount_options): - options += cached_options - except DiskError: - pass - - if not any('subvol=' in x for x in options): - options += f'subvol={self.name}' - - SysCommand(f"mount {self.partition.path} {mountpoint} -o {','.join(options)}") - log(f"{self} has successfully been mounted to {mountpoint}", level=logging.INFO, fg="gray") - - def unmount(self, recurse :bool = True): - SysCommand(f"umount {'-R' if recurse else ''} {self.full_path}") - log(f"Successfully unmounted {self}", level=logging.INFO, fg="gray") |