Send patches - preferably formatted by git format-patch - to patches at archlinux32 dot org.
summaryrefslogtreecommitdiff
path: root/archinstall/lib/disk
diff options
context:
space:
mode:
Diffstat (limited to 'archinstall/lib/disk')
-rw-r--r--archinstall/lib/disk/__init__.py2
-rw-r--r--archinstall/lib/disk/btrfs/__init__.py (renamed from archinstall/lib/disk/btrfs.py)92
-rw-r--r--archinstall/lib/disk/btrfs/btrfs_helpers.py132
-rw-r--r--archinstall/lib/disk/btrfs/btrfspartition.py116
-rw-r--r--archinstall/lib/disk/btrfs/btrfssubvolume.py191
-rw-r--r--archinstall/lib/disk/filesystem.py13
-rw-r--r--archinstall/lib/disk/helpers.py31
-rw-r--r--archinstall/lib/disk/mapperdev.py10
-rw-r--r--archinstall/lib/disk/partition.py120
9 files changed, 580 insertions, 127 deletions
diff --git a/archinstall/lib/disk/__init__.py b/archinstall/lib/disk/__init__.py
index bb6eb815..352d04b9 100644
--- a/archinstall/lib/disk/__init__.py
+++ b/archinstall/lib/disk/__init__.py
@@ -4,4 +4,4 @@ from .blockdevice import BlockDevice
from .filesystem import Filesystem, MBR, GPT
from .partition import *
from .user_guides import *
-from .validators import *
+from .validators import * \ No newline at end of file
diff --git a/archinstall/lib/disk/btrfs.py b/archinstall/lib/disk/btrfs/__init__.py
index 33f59721..84b9c0f6 100644
--- a/archinstall/lib/disk/btrfs.py
+++ b/archinstall/lib/disk/btrfs/__init__.py
@@ -4,44 +4,25 @@ import glob
import logging
import re
from typing import Union, Dict, TYPE_CHECKING, Any, Iterator
-from dataclasses import dataclass
# https://stackoverflow.com/a/39757388/929999
if TYPE_CHECKING:
- from ..installer import Installer
-from .helpers import get_mount_info
-from ..exceptions import DiskError
-from ..general import SysCommand
-from ..output import log
-from ..exceptions import SysCallError
-
-@dataclass
-class BtrfsSubvolume:
- target :str
- source :str
- fstype :str
- name :str
- options :str
- root :bool = False
-
-def get_subvolumes_from_findmnt(struct :Dict[str, Any], index=0) -> Iterator[BtrfsSubvolume]:
- if '[' in struct['source']:
- subvolume = re.findall(r'\[.*?\]', struct['source'])[0][1:-1]
- struct['source'] = struct['source'].replace(f"[{subvolume}]", "")
- yield BtrfsSubvolume(
- target=struct['target'],
- source=struct['source'],
- fstype=struct['fstype'],
- name=subvolume,
- options=struct['options'],
- root=index == 0
- )
- index += 1
-
- for child in struct.get('children', []):
- for item in get_subvolumes_from_findmnt(child, index=index):
- yield item
- index += 1
+ 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 .btrfssubvolume import BtrfsSubvolume as BtrfsSubvolume
+from .btrfspartition import BTRFSPartition as BTRFSPartition
+
+from ..helpers import get_mount_info
+from ...exceptions import DiskError, Deprecated
+from ...general import SysCommand
+from ...output import log
+from ...exceptions import SysCallError
def get_subvolume_info(path :pathlib.Path) -> Dict[str, Any]:
try:
@@ -57,42 +38,6 @@ def get_subvolume_info(path :pathlib.Path) -> Dict[str, Any]:
return result
-def mount_subvolume(installation :Installer, subvolume_location :Union[pathlib.Path, str], force=False) -> bool:
- """
- This function uses mount to mount a subvolume on a given device, at a given location with a given subvolume name.
-
- @installation: archinstall.Installer instance
- @subvolume_location: a localized string or path inside the installation / or /boot for instance without specifying /mnt/boot
- @force: overrides the check for weither or not the subvolume mountpoint is empty or not
-
- This function is DEPRECATED. you can get the same result creating a partition dict like any other partition, and using the standard mount procedure.
- Only change partition['device_instance'].path with the apropriate bind name: real_partition_path[/subvolume_name]
- """
- log("[Deprecated] function btrfs.mount_subvolume is deprecated. See code for alternatives",fg="yellow",level=logging.WARNING)
- 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)
-
- if not target.exists():
- target.mkdir(parents=True)
-
- if glob.glob(str(target / '*')) and force is False:
- raise DiskError(f"Cannot mount subvolume to {target} because it contains data (non-empty folder target)")
-
- log(f"Mounting {target} as a subvolume", level=logging.INFO)
- # Mount the logical volume to the physical structure
- mount_information, mountpoint_device_real_path = get_mount_info(target, traverse=True, return_real_path=True)
- if mountpoint_device_real_path == str(target):
- log(f"Unmounting non-subvolume {mount_information['source']} previously mounted at {target}")
- SysCommand(f"umount {mount_information['source']}")
-
- return SysCommand(f"mount {mount_information['source']} {target} -o subvol=@{subvolume_location}").exit_code == 0
-
def create_subvolume(installation :Installer, subvolume_location :Union[pathlib.Path, str]) -> bool:
"""
This function uses btrfs to create a subvolume.
@@ -132,13 +77,18 @@ def _has_option(option :str,options :list) -> bool:
"""
if not options:
return False
+
for item in options:
if option in item:
return True
+
return False
def manage_btrfs_subvolumes(installation :Installer,
partition :Dict[str, str],) -> list:
+
+ raise Deprecated("Use setup_subvolumes() instead.")
+
from copy import deepcopy
""" we do the magic with subvolumes in a centralized place
parameters:
diff --git a/archinstall/lib/disk/btrfs/btrfs_helpers.py b/archinstall/lib/disk/btrfs/btrfs_helpers.py
new file mode 100644
index 00000000..d529478f
--- /dev/null
+++ b/archinstall/lib/disk/btrfs/btrfs_helpers.py
@@ -0,0 +1,132 @@
+import pathlib
+import logging
+from typing import Optional
+
+from ...exceptions import SysCallError, DiskError
+from ...general import SysCommand
+from ...output import log
+from ..helpers import get_mount_info
+from .btrfssubvolume import BtrfsSubvolume
+
+
+def mount_subvolume(installation, device, name, subvolume_information):
+ # we normalize the subvolume name (getting rid of slash at the start if exists. In our implemenation has no semantic load.
+ # Every subvolume is created from the top of the hierarchy- and simplifies its further use
+ name = name.lstrip('/')
+
+ # renormalize the right hand.
+ mountpoint = subvolume_information.get('mountpoint', None)
+ if not mountpoint:
+ return None
+
+ if type(mountpoint) == str:
+ mountpoint = pathlib.Path(mountpoint)
+
+ installation_target = installation.target
+ if type(installation_target) == str:
+ installation_target = pathlib.Path(installation_target)
+
+ mountpoint = installation_target / mountpoint.relative_to(mountpoint.anchor)
+ mountpoint.mkdir(parents=True, exist_ok=True)
+
+ mount_options = subvolume_information.get('options', [])
+ if not any('subvol=' in x for x in mount_options):
+ mount_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, partition_dict):
+ """
+ Taken from: ..user_guides.py
+
+ partition['btrfs'] = {
+ "subvolumes" : {
+ "@": "/",
+ "@home": "/home",
+ "@log": "/var/log",
+ "@pkg": "/var/cache/pacman/pkg",
+ "@.snapshots": "/.snapshots"
+ }
+ }
+ """
+ log(f"Setting up subvolumes: {partition_dict['btrfs']['subvolumes']}", level=logging.INFO, fg="gray")
+ for name, right_hand in partition_dict['btrfs']['subvolumes'].items():
+ # we normalize the subvolume name (getting rid of slash at the start if exists. In our implemenation has no semantic load.
+ # Every subvolume is created from the top of the hierarchy- and simplifies its further use
+ name = name.lstrip('/')
+
+ # renormalize the right hand.
+ # mountpoint = None
+ subvol_options = []
+
+ match right_hand:
+ # case str(): # backwards-compatability
+ # mountpoint = right_hand
+ case dict():
+ # mountpoint = right_hand.get('mountpoint', None)
+ subvol_options = right_hand.get('options', [])
+
+ # 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 'nodatacow' in subvol_options:
+ 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}")
+ # entry is deleted so nodatacow doesn't propagate to the mount options
+ del subvol_options[subvol_options.index('nodatacow')]
+ # 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 'compress' in subvol_options:
+ 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}")
+ # entry is deleted so compress doesn't propagate to the mount options
+ del subvol_options[subvol_options.index('compress')]
+
+def subvolume_info_from_path(path :pathlib.Path) -> Optional[BtrfsSubvolume]:
+ try:
+ subvolume_name = None
+ 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 BtrfsSubvolume(**{'full_path' : path, 'name' : subvolume_name, **result})
+
+ except SysCallError:
+ pass
+
+ return None
+
+def find_parent_subvolume(path :pathlib.Path, filters=[]):
+ # 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, traverse=True, filters=[*filters, found_mount['target']])
+
+ return subvolume \ No newline at end of file
diff --git a/archinstall/lib/disk/btrfs/btrfspartition.py b/archinstall/lib/disk/btrfs/btrfspartition.py
new file mode 100644
index 00000000..5020133d
--- /dev/null
+++ b/archinstall/lib/disk/btrfs/btrfspartition.py
@@ -0,0 +1,116 @@
+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 .btrfssubvolume import BtrfsSubvolume
+
+class BTRFSPartition(Partition):
+ def __init__(self, *args, **kwargs):
+ Partition.__init__(self, *args, **kwargs)
+
+ def __repr__(self, *args :str, **kwargs :str) -> str:
+ mount_repr = ''
+ if self.mountpoint:
+ mount_repr = f", mounted={self.mountpoint}"
+ elif self.target_mountpoint:
+ mount_repr = f", rel_mountpoint={self.target_mountpoint}"
+
+ if self._encrypted:
+ return f'BTRFSPartition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, parent={self.real_device}, fs={self.filesystem}{mount_repr})'
+ else:
+ return f'BTRFSPartition(path={self.path}, size={self.size}, PARTUUID={self._safe_uuid}, fs={self.filesystem}{mount_repr})'
+
+ @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 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):
+ yield child
+
+ def create_subvolume(self, subvolume :pathlib.Path, installation :Optional['Installer'] = None) -> 'BtrfsSubvolume':
+ """
+ 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 releative 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 absolut 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)")
+ elif subvolinfo := subvolume_info_from_path(subvolume):
+ raise DiskError(f"Destination {subvolume} is already a subvolume: {subvolinfo}")
+
+ SysCommand(f"btrfs subvolume create {subvolume}")
+
+ return subvolume_info_from_path(subvolume) \ No newline at end of file
diff --git a/archinstall/lib/disk/btrfs/btrfssubvolume.py b/archinstall/lib/disk/btrfs/btrfssubvolume.py
new file mode 100644
index 00000000..a96e2a94
--- /dev/null
+++ b/archinstall/lib/disk/btrfs/btrfssubvolume.py
@@ -0,0 +1,191 @@
+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 BtrfsSubvolume:
+ 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
+ # occurance 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 occurance 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") \ No newline at end of file
diff --git a/archinstall/lib/disk/filesystem.py b/archinstall/lib/disk/filesystem.py
index 31929b63..8d8f596e 100644
--- a/archinstall/lib/disk/filesystem.py
+++ b/archinstall/lib/disk/filesystem.py
@@ -65,6 +65,7 @@ class Filesystem:
def load_layout(self, layout :Dict[str, Any]) -> None:
from ..luks import luks2
+ from .btrfs import BTRFSPartition
# If the layout tells us to wipe the drive, we do so
if layout.get('wipe', False):
@@ -142,12 +143,24 @@ class Filesystem:
break
unlocked_device.format(partition['filesystem']['format'], options=format_options)
+
elif partition.get('wipe', False):
if not partition['device_instance']:
raise DiskError(f"Internal error caused us to loose the partition. Please report this issue upstream!")
partition['device_instance'].format(partition['filesystem']['format'], options=format_options)
+ if partition['filesystem']['format'] == 'btrfs':
+ # We upgrade the device instance to a BTRFSPartition if we format it as such.
+ # This is so that we can gain access to more features than otherwise available in Partition()
+ partition['device_instance'] = BTRFSPartition(
+ partition['device_instance'].path,
+ block_device=partition['device_instance'].block_device,
+ encrypted=False,
+ filesystem='btrfs',
+ autodetect_filesystem=False
+ )
+
if partition.get('boot', False):
log(f"Marking partition {partition['device_instance']} as bootable.")
self.set(self.partuuid_to_index(partition['device_instance'].part_uuid), 'boot on')
diff --git a/archinstall/lib/disk/helpers.py b/archinstall/lib/disk/helpers.py
index 0799cd49..99856aad 100644
--- a/archinstall/lib/disk/helpers.py
+++ b/archinstall/lib/disk/helpers.py
@@ -291,11 +291,37 @@ def find_mountpoint(device_path :str) -> Dict[str, Any]:
except SysCallError:
return {}
-def get_mount_info(path :Union[pathlib.Path, str], traverse :bool = False, return_real_path :bool = False) -> Dict[str, Any]:
+def findmnt(path :pathlib.Path, traverse :bool = False, ignore :List = [], recurse :bool = True) -> Dict[str, Any]:
+ for traversal in list(map(str, [str(path)] + list(path.parents))):
+ if traversal in ignore:
+ continue
+
+ try:
+ log(f"Getting mount information for device path {traversal}", level=logging.DEBUG)
+ if (output := SysCommand(f"/usr/bin/findmnt --json {'--submounts' if recurse else ''} {traversal}").decode('UTF-8')):
+ return json.loads(output)
+
+ except SysCallError as error:
+ log(f"Could not get mount information on {path} but continuing and ignoring: {error}", level=logging.INFO, fg="gray")
+ pass
+
+ if not traverse:
+ break
+
+ raise DiskError(f"Could not get mount information for path {path}")
+
+
+def get_mount_info(path :Union[pathlib.Path, str], traverse :bool = False, return_real_path :bool = False, ignore :List = []) -> Dict[str, Any]:
+ import traceback
+
+ log(f"Deprecated: archinstall.get_mount_info(). Use archinstall.findmnt() instead, which does not do any automatic parsing. Please change at:\n{''.join(traceback.format_stack())}")
device_path, bind_path = split_bind_name(path)
output = {}
for traversal in list(map(str, [str(device_path)] + list(pathlib.Path(str(device_path)).parents))):
+ if traversal in ignore:
+ continue
+
try:
log(f"Getting mount information for device path {traversal}", level=logging.DEBUG)
if (output := SysCommand(f'/usr/bin/findmnt --json {traversal}').decode('UTF-8')):
@@ -385,9 +411,8 @@ def get_partitions_in_use(mountpoint :str) -> List[Partition]:
def get_filesystem_type(path :str) -> Optional[str]:
- device_name, bind_name = split_bind_name(path)
try:
- return SysCommand(f"blkid -o value -s TYPE {device_name}").decode('UTF-8').strip()
+ return SysCommand(f"blkid -o value -s TYPE {path}").decode('UTF-8').strip()
except SysCallError:
return None
diff --git a/archinstall/lib/disk/mapperdev.py b/archinstall/lib/disk/mapperdev.py
index 32e3ac9b..67230012 100644
--- a/archinstall/lib/disk/mapperdev.py
+++ b/archinstall/lib/disk/mapperdev.py
@@ -51,11 +51,11 @@ class MapperDev:
raise ValueError(f"Could not convert {self.mappername} to a real dm-crypt device")
@property
- def mountpoint(self) -> Optional[str]:
+ def mountpoint(self) -> Optional[pathlib.Path]:
try:
data = json.loads(SysCommand(f"findmnt --json -R {self.path}").decode())
for filesystem in data['filesystems']:
- return filesystem.get('target')
+ return pathlib.Path(filesystem.get('target'))
except SysCallError as error:
# Not mounted anywhere most likely
@@ -76,8 +76,8 @@ class MapperDev:
@property
def subvolumes(self) -> Iterator['BtrfsSubvolume']:
- from .btrfs import get_subvolumes_from_findmnt
+ from .btrfs import subvolume_info_from_path
for mountpoint in self.mount_information:
- for result in get_subvolumes_from_findmnt(mountpoint):
- yield result \ No newline at end of file
+ if subvolume := subvolume_info_from_path(mountpoint):
+ yield subvolume \ No newline at end of file
diff --git a/archinstall/lib/disk/partition.py b/archinstall/lib/disk/partition.py
index c52ca434..e33c600c 100644
--- a/archinstall/lib/disk/partition.py
+++ b/archinstall/lib/disk/partition.py
@@ -13,7 +13,8 @@ from ..storage import storage
from ..exceptions import DiskError, SysCallError, UnknownFilesystemFormat
from ..output import log
from ..general import SysCommand
-from .btrfs import get_subvolumes_from_findmnt, BtrfsSubvolume
+from .btrfs.btrfs_helpers import subvolume_info_from_path
+from .btrfs.btrfssubvolume import BtrfsSubvolume
class Partition:
def __init__(self,
@@ -96,7 +97,7 @@ class Partition:
try:
data = json.loads(SysCommand(f"findmnt --json -R {self.path}").decode())
for filesystem in data['filesystems']:
- return filesystem.get('target')
+ return pathlib.Path(filesystem.get('target'))
except SysCallError as error:
# Not mounted anywhere most likely
@@ -304,9 +305,26 @@ class Partition:
@property
def subvolumes(self) -> Iterator[BtrfsSubvolume]:
+ from .helpers import findmnt
+
+ def iterate_children_recursively(information):
+ for child in information.get('children', []):
+ if target := child.get('target'):
+ if subvolume := subvolume_info_from_path(pathlib.Path(target)):
+ yield subvolume
+
+ if child.get('children'):
+ for subchild in iterate_children_recursively(child):
+ yield subchild
+
for mountpoint in self.mount_information:
- for result in get_subvolumes_from_findmnt(mountpoint):
- yield result
+ if result := findmnt(pathlib.Path(mountpoint['target'])):
+ for filesystem in result.get('filesystems', []):
+ if subvolume := subvolume_info_from_path(pathlib.Path(mountpoint['target'])):
+ yield subvolume
+
+ for child in iterate_children_recursively(filesystem):
+ yield child
def partprobe(self) -> bool:
try:
@@ -357,7 +375,7 @@ class Partition:
handle = luks2(self, None, None)
return handle.encrypt(self, *args, **kwargs)
- def format(self, filesystem :Optional[str] = None, path :Optional[str] = None, log_formatting :bool = True, options :List[str] = []) -> bool:
+ def format(self, filesystem :Optional[str] = None, path :Optional[str] = None, log_formatting :bool = True, options :List[str] = [], retry :bool = True) -> bool:
"""
Format can be given an overriding path, for instance /dev/null to test
the formatting functionality and in essence the support for the given filesystem.
@@ -379,63 +397,71 @@ class Partition:
if log_formatting:
log(f'Formatting {path} -> {filesystem}', level=logging.INFO)
- if filesystem == 'btrfs':
- options = ['-f'] + options
+ try:
+ if filesystem == 'btrfs':
+ options = ['-f'] + options
- if 'UUID:' not in (mkfs := SysCommand(f"/usr/bin/mkfs.btrfs {' '.join(options)} {path}").decode('UTF-8')):
- raise DiskError(f'Could not format {path} with {filesystem} because: {mkfs}')
- self.filesystem = filesystem
+ if 'UUID:' not in (mkfs := SysCommand(f"/usr/bin/mkfs.btrfs {' '.join(options)} {path}").decode('UTF-8')):
+ raise DiskError(f'Could not format {path} with {filesystem} because: {mkfs}')
+ self.filesystem = filesystem
- elif filesystem == 'vfat':
- options = ['-F32'] + options
+ elif filesystem == 'vfat':
+ options = ['-F32'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
- self.filesystem = filesystem
+ if (handle := SysCommand(f"/usr/bin/mkfs.vfat {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
+ self.filesystem = filesystem
- elif filesystem == 'ext4':
- options = ['-F'] + options
+ elif filesystem == 'ext4':
+ options = ['-F'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.ext4 {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
- self.filesystem = filesystem
+ if (handle := SysCommand(f"/usr/bin/mkfs.ext4 {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
+ self.filesystem = filesystem
- elif filesystem == 'ext2':
- options = ['-F'] + options
+ elif filesystem == 'ext2':
+ options = ['-F'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.ext2 {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f'Could not format {path} with {filesystem} because: {b"".join(handle)}')
- self.filesystem = 'ext2'
+ if (handle := SysCommand(f"/usr/bin/mkfs.ext2 {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f'Could not format {path} with {filesystem} because: {b"".join(handle)}')
+ self.filesystem = 'ext2'
- elif filesystem == 'xfs':
- options = ['-f'] + options
+ elif filesystem == 'xfs':
+ options = ['-f'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.xfs {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
- self.filesystem = filesystem
+ if (handle := SysCommand(f"/usr/bin/mkfs.xfs {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
+ self.filesystem = filesystem
- elif filesystem == 'f2fs':
- options = ['-f'] + options
+ elif filesystem == 'f2fs':
+ options = ['-f'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.f2fs {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
- self.filesystem = filesystem
+ if (handle := SysCommand(f"/usr/bin/mkfs.f2fs {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
+ self.filesystem = filesystem
- elif filesystem == 'ntfs3':
- options = ['-f'] + options
+ elif filesystem == 'ntfs3':
+ options = ['-f'] + options
- if (handle := SysCommand(f"/usr/bin/mkfs.ntfs -Q {' '.join(options)} {path}")).exit_code != 0:
- raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
- self.filesystem = filesystem
+ if (handle := SysCommand(f"/usr/bin/mkfs.ntfs -Q {' '.join(options)} {path}")).exit_code != 0:
+ raise DiskError(f"Could not format {path} with {filesystem} because: {handle.decode('UTF-8')}")
+ self.filesystem = filesystem
- elif filesystem == 'crypto_LUKS':
- # from ..luks import luks2
- # encrypted_partition = luks2(self, None, None)
- # encrypted_partition.format(path)
- self.filesystem = filesystem
+ elif filesystem == 'crypto_LUKS':
+ # from ..luks import luks2
+ # encrypted_partition = luks2(self, None, None)
+ # encrypted_partition.format(path)
+ self.filesystem = filesystem
- else:
- raise UnknownFilesystemFormat(f"Fileformat '{filesystem}' is not yet implemented.")
+ else:
+ raise UnknownFilesystemFormat(f"Fileformat '{filesystem}' is not yet implemented.")
+ except SysCallError as error:
+ log(f"Formatting ran in to an error: {error}", level=logging.WARNING, fg="orange")
+ if retry is True:
+ log(f"Retrying in {storage.get('DISK_TIMEOUTS', 1)} seconds.", level=logging.WARNING, fg="orange")
+ time.sleep(storage.get('DISK_TIMEOUTS', 1))
+
+ return self.format(filesystem, path, log_formatting, options, retry=False)
if get_filesystem_type(path) == 'crypto_LUKS' or get_filesystem_type(self.real_device) == 'crypto_LUKS':
self.encrypted = True