StealthOS/subiquity
StealthOS
/
subiquity
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1383 from dbungert/dispatch-guided-targets 

guided: singledispatch and other refactors
This commit is contained in:
Dan Bungert 2022-08-15 13:22:13 -06:00 committed by GitHub
parent 727a7a8752 7e4db5e769
commit 1a6bfb5bb5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 155 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
subiquity/common/filesystem/gaps.py
View File

@ -70,6 +70,16 @@ class Gap:
usable=self.usable)
return (first_gap, rest_gap)
def within(self):
"""Find the first gap that is contained wholly inside this gap."""
gap_end = self.offset + self.size
for pg in parts_and_gaps(self.device):
if isinstance(pg, Gap):
pg_end = pg.offset + pg.size
if pg.offset >= self.offset and pg_end <= gap_end:
return pg
return None
@functools.singledispatch
def parts_and_gaps(device):
@ -258,17 +268,6 @@ def at_offset(device, offset):
return None
def within(device, gap):
"""Find the first gap that is contained wholly inside the supplied gap."""
gap_end = gap.offset + gap.size
for pg in parts_and_gaps(device):
if isinstance(pg, Gap):
pg_end = pg.offset + pg.size
if pg.offset >= gap.offset and pg_end <= gap_end:
return pg
return None
def after(device, offset):
"""Find the first gap that is after this offset."""
for pg in parts_and_gaps(device):

12
subiquity/common/filesystem/tests/test_gaps.py
View File

@ -121,7 +121,7 @@ class TestWithin(unittest.TestCase):
def test_identity(self):
d = make_disk()
[gap] = gaps.parts_and_gaps(d)
self.assertEqual(gap, gaps.within(d, gap))
self.assertEqual(gap, gap.within())
def test_front_used(self):
m, d = make_model_and_disk(size=200 << 20)
@ -130,7 +130,7 @@ class TestWithin(unittest.TestCase):
[orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
make_partition(m, d, offset=0, size=20 << 20)
[p1, g1, p2, g2] = gaps.parts_and_gaps(d)
self.assertEqual(g1, gaps.within(d, orig_g1))
self.assertEqual(g1, orig_g1.within())
def test_back_used(self):
m, d = make_model_and_disk(size=200 << 20)
@ -139,7 +139,7 @@ class TestWithin(unittest.TestCase):
[orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
make_partition(m, d, offset=80 << 20, size=20 << 20)
[g1, p1, p2, g2] = gaps.parts_and_gaps(d)
self.assertEqual(g1, gaps.within(d, orig_g1))
self.assertEqual(g1, orig_g1.within())
def test_front_and_back_used(self):
m, d = make_model_and_disk(size=200 << 20)
@ -149,7 +149,7 @@ class TestWithin(unittest.TestCase):
make_partition(m, d, offset=0, size=20 << 20)
make_partition(m, d, offset=80 << 20, size=20 << 20)
[p1, g1, p2, p3, g2] = gaps.parts_and_gaps(d)
self.assertEqual(g1, gaps.within(d, orig_g1))
self.assertEqual(g1, orig_g1.within())
def test_multi_gap(self):
m, d = make_model_and_disk(size=200 << 20)
@ -158,7 +158,7 @@ class TestWithin(unittest.TestCase):
[orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
make_partition(m, d, offset=20 << 20, size=20 << 20)
[g1, p1, g2, p2, g3] = gaps.parts_and_gaps(d)
self.assertEqual(g1, gaps.within(d, orig_g1))
self.assertEqual(g1, orig_g1.within())
def test_later_part_of_disk(self):
m, d = make_model_and_disk(size=200 << 20)
@ -167,7 +167,7 @@ class TestWithin(unittest.TestCase):
[orig_g1, p1, orig_g2] = gaps.parts_and_gaps(d)
make_partition(m, d, offset=120 << 20, size=20 << 20)
[g1, p1, g2, p2, g3] = gaps.parts_and_gaps(d)
self.assertEqual(g2, gaps.within(d, orig_g2))
self.assertEqual(g2, orig_g2.within())
class TestAfter(unittest.TestCase):

159
subiquity/server/controllers/filesystem.py
View File

@ -14,6 +14,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import asyncio
import functools
import glob
import json
import logging
@ -41,7 +42,12 @@ from subiquity.common.errorreport import ErrorReportKind
from subiquity.common.filesystem.actions import (
DeviceAction,
)
from subiquity.common.filesystem import boot, gaps, labels, sizes
from subiquity.common.filesystem import (
boot,
gaps,
labels,
sizes,
)
from subiquity.common.filesystem.manipulator import (
FilesystemManipulator,
)
@ -53,6 +59,7 @@ from subiquity.common.types import (
GuidedChoiceV2,
GuidedStorageResponse,
GuidedStorageResponseV2,
GuidedStorageTarget,
GuidedStorageTargetReformat,
GuidedStorageTargetResize,
GuidedStorageTargetUseGap,
@ -65,6 +72,7 @@ from subiquity.common.types import (
from subiquity.models.filesystem import (
align_up,
align_down,
Disk as ModelDisk,
LVM_CHUNK_SIZE,
Raid,
)
@ -138,33 +146,11 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
"autoinstall config did not create needed bootloader "
"partition")
def setup_gap_for_guided(self, target, mode):
if isinstance(target, gaps.Gap):
disk = target.device
gap = target
else:
disk = target
gap = None
if mode is None or mode == 'reformat_disk':
self.reformat(disk, wipe='superblock-recursive')
if DeviceAction.TOGGLE_BOOT in DeviceAction.supported(disk):
self.add_boot_disk(disk)
if gap is None:
return gaps.largest_gap(disk)
else:
# find what's left of the gap after adding boot
gap = gaps.within(disk, gap)
if gap is None:
raise Exception('failed to locate gap after adding boot')
return gap
def guided_direct(self, gap):
spec = dict(fstype="ext4", mount="/")
self.create_partition(device=gap.device, gap=gap, spec=spec)
def guided_direct(self, target, mode=None):
gap = self.setup_gap_for_guided(target, mode)
self.create_partition(
gap.device, gap, dict(fstype="ext4", mount="/"))
def guided_lvm(self, target, mode=None, lvm_options=None):
gap = self.setup_gap_for_guided(target, mode)
def guided_lvm(self, gap, lvm_options=None):
gap_boot, gap_rest = gap.split(sizes.get_bootfs_size(gap.size))
spec = dict(fstype="ext4", mount='/boot')
device = gap.device
@ -205,48 +191,75 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
mount="/",
))
@functools.singledispatchmethod
def start_guided(self, target: GuidedStorageTarget,
disk: ModelDisk) -> gaps.Gap:
"""Setup changes to the disk to prepare the gap that we will be
doing a guided install into."""
raise NotImplementedError(target)
@start_guided.register
def start_guided_reformat(self, target: GuidedStorageTargetReformat,
disk: ModelDisk) -> gaps.Gap:
"""Perform the reformat, and return the resulting gap."""
self.reformat(disk, wipe='superblock-recursive')
return gaps.largest_gap(disk)
@start_guided.register
def start_guided_use_gap(self, target: GuidedStorageTargetUseGap,
disk: ModelDisk) -> gaps.Gap:
"""Lookup the matching model gap."""
return gaps.at_offset(disk, target.gap.offset)
@start_guided.register
def start_guided_resize(self, target: GuidedStorageTargetResize,
disk: ModelDisk) -> gaps.Gap:
"""Perform the resize of the target partition,
and return the resulting gap."""
partition = self.get_partition(disk, target.partition_number)
part_align = disk.alignment_data().part_align
new_size = align_up(target.new_size, part_align)
if new_size > partition.size:
raise Exception(f'Aligned requested size {new_size} too large')
partition.size = new_size
partition.resize = True
# Calculating where that gap will be can be tricky due to alignment
# needs and the possibility that we may be splitting a logical
# partition, which needs an extra 1MiB spacer.
gap = gaps.after(disk, partition.offset)
if gap is None:
pgs = gaps.parts_and_gaps(disk)
raise Exception(f'gap not found after resize, pgs={pgs}')
return gap
def build_lvm_options(self, password):
if password is None:
return None
else:
return {
'encrypt': True,
'luks_options': {
'password': password,
},
}
def guided(self, choice: GuidedChoiceV2):
self.model.guided_configuration = choice
disk = self.model._one(id=choice.target.disk_id)
if isinstance(choice.target, GuidedStorageTargetReformat):
mode = 'reformat_disk'
target = disk
elif isinstance(choice.target, GuidedStorageTargetUseGap):
mode = 'use_gap'
target = gaps.at_offset(disk, choice.target.gap.offset)
elif isinstance(choice.target, GuidedStorageTargetResize):
partition = self.get_partition(
disk, choice.target.partition_number)
part_align = disk.alignment_data().part_align
new_size = align_up(choice.target.new_size, part_align)
if new_size > partition.size:
raise Exception(f'Aligned requested size {new_size} too large')
partition.size = new_size
partition.resize = True
mode = 'use_gap'
# Calculating where that gap will be can be tricky due to alignment
# needs and the possibility that we may be splitting a logical
# partition, which needs an extra 1MiB spacer.
target = gaps.after(disk, partition.offset)
if target is None:
pgs = gaps.parts_and_gaps(disk)
raise Exception(f'gap not found after resize, pgs={pgs}')
else:
raise Exception(f'Unknown guided target {choice.target}')
gap = self.start_guided(choice.target, disk)
if DeviceAction.TOGGLE_BOOT in DeviceAction.supported(disk):
self.add_boot_disk(disk)
# find what's left of the gap after adding boot
gap = gap.within()
if gap is None:
raise Exception('failed to locate gap after adding boot')
if choice.use_lvm:
lvm_options = None
if choice.password is not None:
lvm_options = {
'encrypt': True,
'luks_options': {
'password': choice.password,
},
}
self.guided_lvm(target, mode=mode, lvm_options=lvm_options)
lvm_options = self.build_lvm_options(choice.password)
self.guided_lvm(gap, lvm_options=lvm_options)
else:
self.guided_direct(target, mode=mode)
self.guided_direct(gap)
async def _probe_response(self, wait, resp_cls):
if self._probe_task.task is None or not self._probe_task.task.done():
@ -584,18 +597,15 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
continue
break
def run_guided(self, layout):
def run_autoinstall_guided(self, layout):
name = layout['name']
guided_method = getattr(self, "guided_" + name)
mode = layout.get('mode', 'reformat_disk')
self.validate_layout_mode(mode)
if mode == 'reformat_disk':
match = layout.get("match", {'size': 'largest'})
target = self.model.disk_for_match(self.model.all_disks(), match)
if not target:
raise Exception("autoinstall cannot configure storage "
"- no disk found large enough for install")
disk = self.model.disk_for_match(self.model.all_disks(), match)
target = GuidedStorageTargetReformat(disk_id=disk.id)
elif mode == 'use_gap':
bootable = [d for d in self.model.all_disks()
if boot.can_be_boot_device(d, with_reformatting=False)]
@ -603,12 +613,12 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
if not gap:
raise Exception("autoinstall cannot configure storage "
"- no gap found large enough for install")
# This is not necessarily the exact gap to be used, as the gap size
# may change once add_boot_disk has sorted things out.
target = gap
target = GuidedStorageTargetUseGap(disk_id=gap.device.id, gap=gap)
log.info(f'autoinstall: running guided {name} install in mode {mode} '
f'using {target}')
guided_method(target=target, mode=mode)
use_lvm = name == 'lvm'
self.guided(GuidedChoiceV2(target=target, use_lvm=use_lvm))
def validate_layout_mode(self, mode):
if mode not in ('reformat_disk', 'use_gap'):
@ -618,7 +628,10 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
def convert_autoinstall_config(self, context=None):
log.debug("self.ai_data = %s", self.ai_data)
if 'layout' in self.ai_data:
self.run_guided(self.ai_data['layout'])
if 'config' in self.ai_data:
log.warning("The 'storage' section should not contain both "
"'layout' and 'config', using 'layout'")
self.run_autoinstall_guided(self.ai_data['layout'])
elif 'config' in self.ai_data:
self.model.apply_autoinstall_config(self.ai_data['config'])
self.model.grub = self.ai_data.get('grub')

85
subiquity/server/controllers/tests/test_filesystem.py
View File

@ -94,7 +94,8 @@ class TestGuided(TestCase):
@parameterized.expand(boot_expectations)
def test_guided_direct(self, bootloader, ptable, p1mnt):
self._guided_setup(bootloader, ptable)
self.controller.guided_direct(self.d1)
target = GuidedStorageTargetReformat(disk_id=self.d1.id)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=False))
[d1p1, d1p2] = self.d1.partitions()
self.assertEqual(p1mnt, d1p1.mount)
self.assertEqual('/', d1p2.mount)
@ -102,7 +103,8 @@ class TestGuided(TestCase):
def test_guided_direct_BIOS_MSDOS(self):
self._guided_setup(Bootloader.BIOS, 'msdos')
self.controller.guided_direct(self.d1)
target = GuidedStorageTargetReformat(disk_id=self.d1.id)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=False))
[d1p1] = self.d1.partitions()
self.assertEqual('/', d1p1.mount)
self.assertIsNone(gaps.largest_gap(self.d1))
@ -110,7 +112,8 @@ class TestGuided(TestCase):
@parameterized.expand(boot_expectations)
def test_guided_lvm(self, bootloader, ptable, p1mnt):
self._guided_setup(bootloader, ptable)
self.controller.guided_lvm(self.d1)
target = GuidedStorageTargetReformat(disk_id=self.d1.id)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=True))
[d1p1, d1p2, d1p3] = self.d1.partitions()
self.assertEqual(p1mnt, d1p1.mount)
self.assertEqual('/boot', d1p2.mount)
@ -122,7 +125,8 @@ class TestGuided(TestCase):
def test_guided_lvm_BIOS_MSDOS(self):
self._guided_setup(Bootloader.BIOS, 'msdos')
self.controller.guided_lvm(self.d1)
target = GuidedStorageTargetReformat(disk_id=self.d1.id)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=True))
[d1p1, d1p2] = self.d1.partitions()
self.assertEqual('/boot', d1p1.mount)
[vg] = self.model._all(type='lvm_volgroup')
@ -131,8 +135,8 @@ class TestGuided(TestCase):
self.assertEqual(None, d1p2.mount)
self.assertIsNone(gaps.largest_gap(self.d1))
def _guided_side_by_side(self, bl):
self._guided_setup(bl, 'msdos', storage_version=2)
def _guided_side_by_side(self, bl, ptable):
self._guided_setup(bl, ptable, storage_version=2)
self.controller.add_boot_disk(self.d1)
for p in self.d1._partitions:
p.preserve = True
@ -141,43 +145,60 @@ class TestGuided(TestCase):
self.model._probe_data['blockdev'][p._path()] = {
"ID_PART_ENTRY_TYPE": str(0xef)
}
# create an extended partition,
# and a few other partitions to make it more interesting
# Make it more interesting with other partitions.
# Also create the extended part if needed.
g = gaps.largest_gap(self.d1)
make_partition(self.model, self.d1, preserve=True,
size=10 << 30, offset=g.offset)
g = gaps.largest_gap(self.d1)
make_partition(self.model, self.d1, preserve=True,
flag='extended', size=g.size, offset=g.offset)
g = gaps.largest_gap(self.d1)
make_partition(self.model, self.d1, preserve=True,
flag='logical', size=10 << 30, offset=g.offset)
if ptable == 'msdos':
g = gaps.largest_gap(self.d1)
make_partition(self.model, self.d1, preserve=True,
flag='extended', size=g.size, offset=g.offset)
g = gaps.largest_gap(self.d1)
make_partition(self.model, self.d1, preserve=True,
flag='logical', size=10 << 30, offset=g.offset)
@parameterized.expand(bootloaders)
def test_guided_direct_side_by_side_logical(self, bl):
self._guided_side_by_side(bl)
@parameterized.expand(
[(bl, pt, flag)
for bl in list(Bootloader)
for pt, flag in (
('msdos', 'logical'),
('gpt', None)
)]
)
def test_guided_direct_side_by_side(self, bl, pt, flag):
self._guided_side_by_side(bl, pt)
parts_before = self.d1._partitions.copy()
g = gaps.largest_gap(self.d1)
self.controller.guided_direct(g, mode='use_gap')
gap = gaps.largest_gap(self.d1)
target = GuidedStorageTargetUseGap(disk_id=self.d1.id, gap=gap)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=False))
parts_after = gaps.parts_and_gaps(self.d1)[:-1]
self.assertEqual(parts_before, parts_after)
p6 = gaps.parts_and_gaps(self.d1)[-1]
self.assertEqual('/', p6.mount)
self.assertEqual('logical', p6.flag)
p = gaps.parts_and_gaps(self.d1)[-1]
self.assertEqual('/', p.mount)
self.assertEqual(flag, p.flag)
@parameterized.expand(bootloaders)
def test_guided_lvm_side_by_side_logical(self, bl):
self._guided_side_by_side(bl)
@parameterized.expand(
[(bl, pt, flag)
for bl in list(Bootloader)
for pt, flag in (
('msdos', 'logical'),
('gpt', None)
)]
)
def test_guided_lvm_side_by_side(self, bl, pt, flag):
self._guided_side_by_side(bl, pt)
parts_before = self.d1._partitions.copy()
g = gaps.largest_gap(self.d1)
self.controller.guided_lvm(g, mode='use_gap')
gap = gaps.largest_gap(self.d1)
target = GuidedStorageTargetUseGap(disk_id=self.d1.id, gap=gap)
self.controller.guided(GuidedChoiceV2(target=target, use_lvm=True))
parts_after = gaps.parts_and_gaps(self.d1)[:-2]
self.assertEqual(parts_before, parts_after)
p6, p7 = gaps.parts_and_gaps(self.d1)[-2:]
self.assertEqual('/boot', p6.mount)
self.assertEqual('logical', p6.flag)
self.assertEqual(None, p7.mount)
self.assertEqual('logical', p7.flag)
p_boot, p_data = gaps.parts_and_gaps(self.d1)[-2:]
self.assertEqual('/boot', p_boot.mount)
self.assertEqual(flag, p_boot.flag)
self.assertEqual(None, p_data.mount)
self.assertEqual(flag, p_data.flag)
class TestLayout(TestCase):