almalinux/kiwi-el8
14
0
Fork 0
Code Issues Pull Requests Releases Activity

Add support for part clones to the Disk interface

Browse Source 
The Disk class provides methods to create partition(s)
and map names according to its scope and independent of
the actual partition tools. For example: create_root_partition().
This commit adds an additional optional clone parameter to
all methods for which we want to allow partition clones
This commit is contained in:
Marcus Schäfer 2022-04-10 16:52:05 +02:00
parent fe5f053f9d
commit 69c5495b79
No known key found for this signature in database
GPG Key ID: A16C1128698C8CAC
5 changed files with 183 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
kiwi/storage/disk.py
View File

@ -19,7 +19,7 @@ import os
import logging
from collections import OrderedDict
from typing import (
Dict, NamedTuple
Dict, NamedTuple, Tuple
)
# project
@ -33,6 +33,7 @@ from kiwi.exceptions import KiwiCustomPartitionConflictError
ptable_entry_type = NamedTuple(
'ptable_entry_type', [
('mbsize', int),
('clone', int),
('partition_name', str),
('partition_type', str),
('mountpoint', str),
@ -145,23 +146,32 @@ class Disk(DeviceProvider):
raise KiwiCustomPartitionConflictError(
f'Cannot use reserved table entry name: {map_name!r}'
)
id_name = f'kiwi_{map_name.title()}Part'
entry = table_entries[map_name]
if entry.clone:
self._create_clones(
map_name, entry.clone, entry.partition_type,
format(entry.mbsize)
)
id_name = f'kiwi_{map_name.title()}Part'
self.partitioner.create(
entry.partition_name, entry.mbsize, entry.partition_type
)
self._add_to_map(map_name)
self._add_to_public_id_map(id_name)
def create_root_partition(self, mbsize: str):
def create_root_partition(self, mbsize: str, clone: int = 0):
"""
Create root partition
Populates kiwi_RootPart(id) and kiwi_BootPart(id) if no extra
boot partition is requested
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
:param int clone: create [clone] cop(y/ies) of the root partition
"""
(mbsize, mbsize_clone) = self._parse_size(mbsize)
if clone:
self._create_clones('root', clone, 't.linux', mbsize_clone)
self.partitioner.create('p.lxroot', mbsize, 't.linux')
self._add_to_map('root')
self._add_to_public_id_map('kiwi_RootPart')
@ -170,19 +180,23 @@ class Disk(DeviceProvider):
if 'kiwi_BootPart' not in self.public_partition_id_map:
self._add_to_public_id_map('kiwi_BootPart')
def create_root_lvm_partition(self, mbsize: str):
def create_root_lvm_partition(self, mbsize: str, clone: int = 0):
"""
Create root partition for use with LVM
Populates kiwi_RootPart(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
:param int clone: create [clone] cop(y/ies) of the lvm roo partition
"""
(mbsize, mbsize_clone) = self._parse_size(mbsize)
if clone:
self._create_clones('root', clone, 't.lvm', mbsize_clone)
self.partitioner.create('p.lxlvm', mbsize, 't.lvm')
self._add_to_map('root')
self._add_to_public_id_map('kiwi_RootPart')
def create_root_raid_partition(self, mbsize: str):
def create_root_raid_partition(self, mbsize: str, clone: int = 0):
"""
Create root partition for use with MD Raid
@ -190,14 +204,18 @@ class Disk(DeviceProvider):
as the default raid device node at boot time which is
configured to be kiwi_RaidDev(/dev/mdX)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
:param int clone: create [clone] cop(y/ies) of the raid root partition
"""
(mbsize, mbsize_clone) = self._parse_size(mbsize)
if clone:
self._create_clones('root', clone, 't.raid', mbsize_clone)
self.partitioner.create('p.lxraid', mbsize, 't.raid')
self._add_to_map('root')
self._add_to_public_id_map('kiwi_RootPart')
self._add_to_public_id_map('kiwi_RaidPart')
def create_root_readonly_partition(self, mbsize: str):
def create_root_readonly_partition(self, mbsize: str, clone: int = 0):
"""
Create root readonly partition for use with overlayfs
@ -206,20 +224,28 @@ class Disk(DeviceProvider):
should be the size of the squashfs filesystem in order to
avoid wasting disk space
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
:param int clone: create [clone] cop(y/ies) of the ro root partition
"""
(mbsize, mbsize_clone) = self._parse_size(mbsize)
if clone:
self._create_clones('root', clone, 't.linux', mbsize_clone)
self.partitioner.create('p.lxreadonly', mbsize, 't.linux')
self._add_to_map('readonly')
self._add_to_public_id_map('kiwi_ROPart')
def create_boot_partition(self, mbsize: str):
def create_boot_partition(self, mbsize: str, clone: int = 0):
"""
Create boot partition
Populates kiwi_BootPart(id)
Populates kiwi_BootPart(id) and optional kiwi_BootPartClone(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
:param int clone: create [clone] cop(y/ies) of the boot partition
"""
(mbsize, mbsize_clone) = self._parse_size(mbsize)
if clone:
self._create_clones('boot', clone, 't.linux', mbsize_clone)
self.partitioner.create('p.lxboot', mbsize, 't.linux')
self._add_to_map('boot')
self._add_to_public_id_map('kiwi_BootPart')
@ -230,8 +256,9 @@ class Disk(DeviceProvider):
Populates kiwi_PrepPart(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
"""
(mbsize, _) = self._parse_size(mbsize)
self.partitioner.create('p.prep', mbsize, 't.prep')
self._add_to_map('prep')
self._add_to_public_id_map('kiwi_PrepPart')
@ -242,8 +269,9 @@ class Disk(DeviceProvider):
Populates kiwi_SparePart(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
"""
(mbsize, _) = self._parse_size(mbsize)
self.partitioner.create('p.spare', mbsize, 't.linux')
self._add_to_map('spare')
self._add_to_public_id_map('kiwi_SparePart')
@ -254,8 +282,9 @@ class Disk(DeviceProvider):
Populates kiwi_SwapPart(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
"""
(mbsize, _) = self._parse_size(mbsize)
self.partitioner.create('p.swap', mbsize, 't.swap')
self._add_to_map('swap')
self._add_to_public_id_map('kiwi_SwapPart')
@ -266,8 +295,9 @@ class Disk(DeviceProvider):
Populates kiwi_BiosGrub(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
"""
(mbsize, _) = self._parse_size(mbsize)
self.partitioner.create('p.legacy', mbsize, 't.csm')
self._add_to_map('efi_csm')
self._add_to_public_id_map('kiwi_BiosGrub')
@ -278,8 +308,9 @@ class Disk(DeviceProvider):
Populates kiwi_EfiPart(id)
:param str mbsize: partition size NumberString or 'all_free'
:param str mbsize: partition size string
"""
(mbsize, _) = self._parse_size(mbsize)
self.partitioner.create('p.UEFI', mbsize, 't.efi')
self._add_to_map('efi')
self._add_to_public_id_map('kiwi_EfiPart')
@ -377,6 +408,60 @@ class Disk(DeviceProvider):
sorted(self.public_partition_id_map.items())
)
def _create_clones(
self, name: str, clone: int, type_flag: str, mbsize: str
) -> None:
"""
Create [clone] cop(y/ies) of the given partition name
The name of a clone partition uses the following name policy:
* {name}clone{id} for the partition name
* kiwi_{name}PartClone{id} for the kiwi map name
:param str name: basename to use for clone partition names
:param int clone: number of clones, >= 1
:param str type_flag: partition type name
:param str mbsize: partition size string
"""
for clone_id in range(1, clone + 1):
self.partitioner.create(
f'p.lx{name}clone{clone_id}', mbsize, type_flag
)
self._add_to_map(f'{name}clone{clone_id}')
self._add_to_public_id_map(f'kiwi_{name}PartClone{clone_id}')
def _parse_size(self, value: str) -> Tuple[str, str]:
"""
parse size value. This can be one of the following
* A number_string
* The string named: 'all_free'
* The string formatted as:
clone:{number_string_origin}:{number_string_clone}
The method returns a tuple for size and optional clone size
If no clone size exists both tuple values are the same
The given number_string for the size of the partition is
passed along to the actually used partitioner object and
expected to be valid there. In case invalid size information
is passed to the partitioner an exception will be raised
in the scope of the partitioner interface and the selected
partitioner class
:param str value: size value
:return: Tuple of strings
:rtype: tuple
"""
if not format(value).startswith('clone:'):
return (value, value)
else:
size_list = value.split(':')
return (size_list[1], size_list[2])
def _add_to_public_id_map(self, name, value=None):
if not value:
value = self.partitioner.get_id()

6
kiwi/xml_state.py
View File

@ -1418,6 +1418,7 @@ class XMLState:
{
'NAME': ptable_entry_type(
mbsize=int,
clone=int,
partition_name=str,
partition_type=str,
mountpoint=str,
@ -1436,6 +1437,11 @@ class XMLState:
partition_name = partition.get_partition_name() or f'p.lx{name}'
partitions[name] = ptable_entry_type(
mbsize=self._to_mega_byte(partition.get_size()),
# There is currently no clone attribute in the <partition>
# element. This will be added on completion of the partition
# clone feature. The internal API structure however, already
# knows about the capability
clone=0,
partition_name=partition_name,
partition_type=partition.get_partition_type() or 't.linux',
mountpoint=partition.get_mountpoint(),

1
test/unit/builder/disk_test.py
View File

@ -803,6 +803,7 @@ class TestDiskBuilder:
self.disk_builder.custom_partitions = {
'var': ptable_entry_type(
mbsize=100,
clone=0,
partition_name='p.lxvar',
partition_type='t.linux',
mountpoint='/var',

119
test/unit/storage/disk_test.py
View File

@ -53,90 +53,99 @@ class TestDisk:
self.storage_provider.is_loop.called_once_with()
def test_create_root_partition(self):
self.disk.create_root_partition(100)
self.partitioner.create.assert_called_once_with(
'p.lxroot', 100, 't.linux'
)
self.disk.create_root_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.linux'),
call('p.lxroot', '100', 't.linux')
]
def test_create_root_which_is_also_boot_partition(self):
self.disk.create_root_partition(200)
self.disk.create_root_partition('200')
self.partitioner.create.assert_called_once_with(
'p.lxroot', 200, 't.linux'
'p.lxroot', '200', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_BootPart'] == 1
def test_create_root_which_is_also_read_write_partition(self):
self.disk.public_partition_id_map['kiwi_ROPart'] = 1
self.disk.create_root_partition(200)
self.disk.create_root_partition('200')
self.partitioner.create.assert_called_once_with(
'p.lxroot', 200, 't.linux'
'p.lxroot', '200', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_RWPart'] == 1
def test_create_root_lvm_partition(self):
self.disk.create_root_lvm_partition(100)
self.partitioner.create.assert_called_once_with(
'p.lxlvm', 100, 't.lvm'
)
self.disk.create_root_lvm_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.lvm'),
call('p.lxlvm', '100', 't.lvm')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
def test_create_root_raid_partition(self):
self.disk.create_root_raid_partition(100)
self.partitioner.create.assert_called_once_with(
'p.lxraid', 100, 't.raid'
)
self.disk.create_root_raid_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.raid'),
call('p.lxraid', '100', 't.raid')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_RootPart'] == 1
assert self.disk.public_partition_id_map['kiwi_RaidPart'] == 1
def test_create_root_readonly_partition(self):
self.disk.create_root_readonly_partition(100)
self.partitioner.create.assert_called_once_with(
'p.lxreadonly', 100, 't.linux'
)
self.disk.create_root_readonly_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxrootclone1', '100', 't.linux'),
call('p.lxreadonly', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_rootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_ROPart'] == 1
def test_create_boot_partition(self):
self.disk.create_boot_partition(100)
self.partitioner.create.assert_called_once_with(
'p.lxboot', 100, 't.linux'
)
self.disk.create_boot_partition('100', 1)
assert self.partitioner.create.call_args_list == [
call('p.lxbootclone1', '100', 't.linux'),
call('p.lxboot', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_bootPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_BootPart'] == 1
def test_create_efi_csm_partition(self):
self.disk.create_efi_csm_partition(100)
self.disk.create_efi_csm_partition('100')
self.partitioner.create.assert_called_once_with(
'p.legacy', 100, 't.csm'
'p.legacy', '100', 't.csm'
)
assert self.disk.public_partition_id_map['kiwi_BiosGrub'] == 1
def test_create_efi_partition(self):
self.disk.create_efi_partition(100)
self.disk.create_efi_partition('100')
self.partitioner.create.assert_called_once_with(
'p.UEFI', 100, 't.efi'
'p.UEFI', '100', 't.efi'
)
assert self.disk.public_partition_id_map['kiwi_EfiPart'] == 1
def test_create_spare_partition(self):
self.disk.create_spare_partition(42)
self.disk.create_spare_partition('42')
self.partitioner.create.assert_called_once_with(
'p.spare', 42, 't.linux'
'p.spare', '42', 't.linux'
)
assert self.disk.public_partition_id_map['kiwi_SparePart'] == 1
def test_create_swap_partition(self):
self.disk.create_swap_partition(42)
self.disk.create_swap_partition('42')
self.partitioner.create.assert_called_once_with(
'p.swap', 42, 't.swap'
'p.swap', '42', 't.swap'
)
assert self.disk.public_partition_id_map['kiwi_SwapPart'] == 1
@patch('kiwi.storage.disk.Command.run')
def test_create_prep_partition(self, mock_command):
self.disk.create_prep_partition(8)
self.disk.create_prep_partition('8')
self.partitioner.create.assert_called_once_with(
'p.prep', 8, 't.prep'
'p.prep', '8', 't.prep'
)
assert self.disk.public_partition_id_map['kiwi_PrepPart'] == 1
@ -144,7 +153,8 @@ class TestDisk:
def test_create_custom_partitions(self, mock_command):
table_entries = {
'var': ptable_entry_type(
mbsize=100,
mbsize='100',
clone=2,
partition_name='p.lxvar',
partition_type='t.linux',
mountpoint='/var',
@ -152,15 +162,20 @@ class TestDisk:
)
}
self.disk.create_custom_partitions(table_entries)
self.partitioner.create.assert_called_once_with(
'p.lxvar', 100, 't.linux'
)
assert self.partitioner.create.call_args_list == [
call('p.lxvarclone1', '100', 't.linux'),
call('p.lxvarclone2', '100', 't.linux'),
call('p.lxvar', '100', 't.linux')
]
assert self.disk.public_partition_id_map['kiwi_varPartClone1'] == 1
assert self.disk.public_partition_id_map['kiwi_varPartClone2'] == 1
assert self.disk.public_partition_id_map['kiwi_VarPart'] == 1
def test_create_custom_partitions_reserved_name(self):
table_entries = {
'root': ptable_entry_type(
mbsize=100,
mbsize='100',
clone=0,
partition_name='p.lxroot',
partition_type='t.linux',
mountpoint='/',
@ -172,14 +187,14 @@ class TestDisk:
@patch('kiwi.storage.disk.Command.run')
def test_device_map_efi_partition(self, mock_command):
self.disk.create_efi_partition(100)
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/mapper/loop0p1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_device_map_prep_partition(self, mock_command):
self.disk.create_prep_partition(8)
self.disk.create_prep_partition('8')
self.disk.map_partitions()
assert self.disk.partition_map == {'prep': '/dev/mapper/loop0p1'}
self.disk.is_mapped = False
@ -190,7 +205,7 @@ class TestDisk:
self.storage_provider.get_device = mock.Mock(
return_value='/dev/sda'
)
self.disk.create_efi_partition(100)
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/sda1'}
self.disk.is_mapped = False
@ -201,27 +216,27 @@ class TestDisk:
self.storage_provider.get_device = mock.Mock(
return_value='/dev/c0d0'
)
self.disk.create_efi_partition(100)
self.disk.create_efi_partition('100')
self.disk.map_partitions()
assert self.disk.partition_map == {'efi': '/dev/c0d0p1'}
self.disk.is_mapped = False
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_boot_partition(self, mock_command):
self.disk.create_boot_partition(100)
self.disk.create_root_partition(100)
self.disk.create_boot_partition('100')
self.disk.create_root_partition('100')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_root_partition(self, mock_command):
self.disk.create_root_partition(100)
self.disk.create_root_partition('100')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@patch('kiwi.storage.disk.Command.run')
def test_activate_boot_partition_is_prep_partition(self, mock_command):
self.disk.create_prep_partition(8)
self.disk.create_prep_partition('8')
self.disk.activate_boot_partition()
self.partitioner.set_flag(1, 'f.active')
@ -300,3 +315,13 @@ class TestDisk:
def test_create_mbr(self):
self.disk.create_mbr()
self.partitioner.set_mbr.assert_called_once_with()
def test_parse_size(self):
(size, _) = self.disk._parse_size('100')
assert size == '100'
(size, clone_size) = self.disk._parse_size('all_free')
assert size == 'all_free'
assert clone_size == 'all_free'
(size, clone_size) = self.disk._parse_size('clone:100:all_free')
assert size == '100'
assert clone_size == 'all_free'

1
test/unit/xml_state_test.py
View File

@ -330,6 +330,7 @@ class TestXMLState:
assert state.get_partitions() == {
'var': ptable_entry_type(
mbsize=100,
clone=0,
partition_name='p.lxvar',
partition_type='t.linux',
mountpoint='/var',