1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library 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
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
"""RAID related functions"""
from pathlib import Path
from shutil import copy
from dataclasses import dataclass
from vyos.utils.process import cmd
from vyos.system import disk
@dataclass
class RaidDetails:
"""RAID type"""
name: str
level: str
members: list[str]
disks: list[disk.DiskDetails]
def raid_create(raid_members: list[str],
raid_name: str = 'md0',
raid_level: str = 'raid1') -> None:
"""Create a RAID array
Args:
raid_name (str): a name of array (data, backup, test, etc.)
raid_members (list[str]): a list of array members
raid_level (str, optional): an array level. Defaults to 'raid1'.
"""
raid_devices_num: int = len(raid_members)
raid_members_str: str = ' '.join(raid_members)
if Path('/sys/firmware/efi').exists():
for part in raid_members:
drive: str = disk.partition_parent(part)
command: str = f'sgdisk --typecode=3:A19D880F-05FC-4D3B-A006-743F0F84911E {drive}'
cmd(command)
else:
for part in raid_members:
drive: str = disk.partition_parent(part)
command: str = f'sgdisk --typecode=3:A19D880F-05FC-4D3B-A006-743F0F84911E {drive}'
cmd(command)
for part in raid_members:
command: str = f'mdadm --zero-superblock {part}'
cmd(command)
command: str = f'mdadm --create /dev/{raid_name} -R --metadata=1.0 \
--raid-devices={raid_devices_num} --level={raid_level} \
{raid_members_str}'
cmd(command)
raid = RaidDetails(
name = f'/dev/{raid_name}',
level = raid_level,
members = raid_members,
disks = [disk.from_partition(m) for m in raid_members]
)
return raid
def update_initramfs() -> None:
"""Update initramfs"""
mdadm_script = '/etc/initramfs-tools/scripts/local-top/mdadm'
copy('/usr/share/initramfs-tools/scripts/local-block/mdadm', mdadm_script)
p = Path(mdadm_script)
p.write_text(p.read_text().replace('$((COUNT + 1))', '20'))
command: str = 'update-initramfs -u'
cmd(command)
def update_default(target_dir: str) -> None:
"""Update /etc/default/mdadm to start MD monitoring daemon at boot
"""
source_mdadm_config = '/etc/default/mdadm'
target_mdadm_config = Path(target_dir).joinpath('/etc/default/mdadm')
target_mdadm_config_dir = Path(target_mdadm_config).parent
Path.mkdir(target_mdadm_config_dir, parents=True, exist_ok=True)
s = Path(source_mdadm_config).read_text().replace('START_DAEMON=false',
'START_DAEMON=true')
Path(target_mdadm_config).write_text(s)
def get_uuid(device: str) -> str:
"""Get UUID of a device"""
command: str = f'tune2fs -l {device}'
l = cmd(command).splitlines()
uuid = next((x for x in l if x.startswith('Filesystem UUID')), '')
return uuid.split(':')[1].strip() if uuid else ''
def get_uuids(raid_details: RaidDetails) -> tuple[str]:
"""Get UUIDs of RAID members
Args:
raid_name (str): a name of array (data, backup, test, etc.)
Returns:
tuple[str]: root_disk uuid, root_md uuid
"""
raid_name: str = raid_details.name
root_partition: str = raid_details.members[0]
uuid_root_disk: str = get_uuid(root_partition)
uuid_root_md: str = get_uuid(raid_name)
return uuid_root_disk, uuid_root_md
|