Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

WIP: Adapt to latest snapper #2697

Draft
wants to merge 2 commits into
base: main
Choose a base branch
from
Draft

WIP: Adapt to latest snapper #2697

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a299a23
Adding the ChrootManager class
davidcassany Jan 16, 2025
795f0bf
WIP: Adapt to latest snapper
davidcassany Dec 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 119 additions & 0 deletions kiwi/chroot_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,119 @@
# Copyright (c) 2025 SUSE Software Solutions Germany GmbH. All rights reserved.
#
# This file is part of kiwi.
#
# kiwi is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# kiwi is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
import os
import logging
from typing import (
List, Optional
)

# project
from kiwi.mount_manager import MountManager
from kiwi.command import (
Command, CommandT, MutableMapping
)
from kiwi.exceptions import (
KiwiUmountBusyError
)

log = logging.getLogger('kiwi')


class ChrootManager:
"""
**Implements methods for setting and unsetting a chroot environment**

The caller is responsible for cleaning up bind mounts if the ChrootManager
is used as is.

The class also supports to be used as a context manager, where any bind or kernel
filesystem mount is unmounted once the context manager's with block is left

* :param string root_dir: path to change the root to
* :param list binds: current root paths to bind to the chrooted path
"""
def __init__(self, root_dir: str, binds: List[str] = []):
self.root_dir = root_dir
self.mounts: List[MountManager] = []
for bind in binds:
self.mounts.append(MountManager(
device=bind, mountpoint=os.path.normpath(root_dir + bind)
))

def __enter__(self) -> "ChrootManager":
try:
self.mount()
except Exception as e:
self.umount()
raise KiwiChrootEnterError(e)
return self

def __exit__(self, exc_type, exc_value, traceback) -> None:
self.umount()

def mount(self) -> None:
"""
Mounts binds to the chroot path
"""
for mnt in self.mounts:
mnt.bind_mount()

def umount(self) -> None:
"""
Unmounts all binds from the chroot path

If any unmount raises a KiwiUmountBusyError this is trapped
and kept until the iteration over all bind mounts is over.
"""
errors = []
for mnt in reversed(self.mounts):
try:
mnt.umount()
except KiwiUmountBusyError as e:
errors.append(e)

if errors:
raise KiwiUmountBusyError(errors)

def run(
self, command: List[str],
custom_env: Optional[MutableMapping[str, str]] = None,
raise_on_error: bool = True, stderr_to_stdout: bool = False,
raise_on_command_not_found: bool = True
) -> Optional[CommandT]:
"""
This is a wrapper for Command.run method but pre-appending the
chroot call at the command list

:param list command: command and arguments
:param dict custom_env: custom os.environ
:param bool raise_on_error: control error behaviour
:param bool stderr_to_stdout: redirects stderr to stdout

:return:
Contains call results in command type

.. code:: python

CommandT(output='string', error='string', returncode=int)

:rtype: CommandT
"""
chroot_cmd = ['chroot', self.root_dir]
chroot_cmd = chroot_cmd + command
return Command.run(chroot_cmd)

113 changes: 29 additions & 84 deletions kiwi/volume_manager/btrfs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from kiwi.command import Command
from kiwi.volume_manager.base import VolumeManagerBase
from kiwi.mount_manager import MountManager
from kiwi.chroot_manager import ChrootManager
from kiwi.storage.mapped_device import MappedDevice
from kiwi.filesystem import FileSystem
from kiwi.utils.sync import DataSync
Expand Down Expand Up @@ -135,21 +136,14 @@ def setup(self, name=None):
['btrfs', 'subvolume', 'create', root_volume]
)
if self.custom_args['root_is_snapper_snapshot']:
snapshot_volume = self.mountpoint + \
f'/{self.root_volume_name}/.snapshots'
Command.run(
['btrfs', 'subvolume', 'create', snapshot_volume]
)
os.chmod(snapshot_volume, 0o700)
Path.create(snapshot_volume + '/1')
snapshot = self.mountpoint + \
f'/{self.root_volume_name}/.snapshots/1/snapshot'
Command.run(
['btrfs', 'subvolume', 'create', snapshot]
)
self._set_default_volume(
f'{self.root_volume_name}/.snapshots/1/snapshot'
)
with ChrootManager(
self.root_dir, binds=[self.mountpoint]
) as chroot:
chroot.run([
'/usr/lib/snapper/installation-helper', '--root-prefix',
os.path.join(self.mountpoint, self.root_volume_name),
'--step', 'filesystem'
])
snapshot = self.mountpoint + \
f'/{self.root_volume_name}/.snapshots/1/snapshot'
# Mount /{some-name}/.snapshots as /.snapshots inside the root
Expand Down Expand Up @@ -413,19 +407,24 @@ def sync_data(self, exclude=None):
"""
if self.toplevel_mount:
sync_target = self.get_mountpoint()
if self.custom_args['root_is_snapper_snapshot']:
self._create_snapshot_info(
''.join(
[
self.mountpoint,
f'/{self.root_volume_name}/.snapshots/1/info.xml'
]
)
)
data = DataSync(self.root_dir, sync_target)
data.sync_data(
options=Defaults.get_sync_options(), exclude=exclude
)
if self.custom_args['root_is_snapper_snapshot']:
root_prefix = os.path.join(
self.mountpoint,
f'{self.root_volume_name}/.snapshots/1/snapshot'
Comment on lines +414 to +417
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we have snapper tell us what this is? This is still an assumption of how snapper works rather than snapper telling us stuff.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Indeed this is a though one and I fully see your point. However I am not sure how this could be actually solved, all this refactor is based on the two step installation helper introduced in snapper 0.12.1 (see openSUSE/snapper#944). From a snapper PoV this could be solved with docs, I mean, this is just constant in the installation helper. I´d say it always creates a subvolume with ID=1 and the snapshots structure is also a constant. Gonna have a look if there is anything from snapper to provide the path for a given snapshot or something similar.

)
snapshots_prefix = os.path.join(root_prefix, '.snapshots')
with ChrootManager(
self.root_dir, binds=[root_prefix, snapshots_prefix]
) as chroot:
chroot.run([
'/usr/lib/snapper/installation-helper', '--root-prefix',
root_prefix, '--step', 'config', '--description',
'first root filesystem'
])
if self.custom_args['quota_groups'] and \
self.custom_args['root_is_snapper_snapshot']:
self._create_snapper_quota_configuration()
Expand Down Expand Up @@ -503,76 +502,22 @@ def _set_default_volume(self, default_volume):
'Failed to find btrfs volume: %s' % default_volume
)

def _xml_pretty(self, toplevel_element):
xml_data_unformatted = ElementTree.tostring(
toplevel_element, 'utf-8'
)
xml_data_domtree = minidom.parseString(xml_data_unformatted)
return xml_data_domtree.toprettyxml(indent=" ")

def _create_snapper_quota_configuration(self):
root_path = os.sep.join(
[
self.mountpoint,
f'{self.root_volume_name}/.snapshots/1/snapshot'
]
)
snapper_default_conf = Defaults.get_snapper_config_template_file(
root_path
# snapper requires an extra parent qgroup to operate with quotas
Command.run(
['btrfs', 'qgroup', 'create', '1/0', self.mountpoint]
Comment on lines +512 to +514
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is again something that snapper should do rather than us.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not so sure about this, I don´t know enough about this use case for a proper opinion, but at a glance I am not convinced this is on snapper responsibilities. Snapper only manages subvolumes in a CRUD fashion and it expects a btrfs filesystem there ready to be used. We are just telling to snapper which quota group to use on the existing filesystem.

)
if snapper_default_conf:
# snapper requires an extra parent qgroup to operate with quotas
Command.run(
['btrfs', 'qgroup', 'create', '1/0', self.mountpoint]
)
config_file = self._set_snapper_sysconfig_file(root_path)
if not os.path.exists(config_file):
shutil.copyfile(snapper_default_conf, config_file)
Command.run([
'chroot', root_path, 'snapper', '--no-dbus', 'set-config',
'QGROUP=1/0'
with ChrootManager(root_path) as chroot:
chroot.run([
'snapper', '--no-dbus', 'set-config', 'QGROUP=1/0'
])

@staticmethod
def _set_snapper_sysconfig_file(root_path):
sysconf_file = SysConfig(
os.sep.join([root_path, 'etc/sysconfig/snapper'])
)
if not sysconf_file.get('SNAPPER_CONFIGS') or \
len(sysconf_file['SNAPPER_CONFIGS'].strip('\"')) == 0:

sysconf_file['SNAPPER_CONFIGS'] = '"root"'
sysconf_file.write()
elif len(sysconf_file['SNAPPER_CONFIGS'].split()) > 1:
raise KiwiVolumeManagerSetupError(
'Unsupported SNAPPER_CONFIGS value: {0}'.format(
sysconf_file['SNAPPER_CONFIGS']
)
)
return os.sep.join([
root_path, 'etc/snapper/configs',
sysconf_file['SNAPPER_CONFIGS'].strip('\"')]
)

def _create_snapshot_info(self, filename):
date_info = datetime.datetime.now()
snapshot = ElementTree.Element('snapshot')

snapshot_type = ElementTree.SubElement(snapshot, 'type')
snapshot_type.text = 'single'

snapshot_number = ElementTree.SubElement(snapshot, 'num')
snapshot_number.text = '1'

snapshot_description = ElementTree.SubElement(snapshot, 'description')
snapshot_description.text = 'first root filesystem'

snapshot_date = ElementTree.SubElement(snapshot, 'date')
snapshot_date.text = date_info.strftime("%Y-%m-%d %H:%M:%S")

with open(filename, 'w') as snapshot_info_file:
snapshot_info_file.write(self._xml_pretty(snapshot))

def __exit__(self, exc_type, exc_value, traceback):
if self.toplevel_mount:
self.umount_volumes()
Loading