File diff suppressed because it is too large
Load Diff
@@ -1,142 +1,34 @@
|
|||||||
import json
|
def get_instance_name(machine_name: str) -> str:
|
||||||
import logging
|
return f"{machine_name}-single-disk"
|
||||||
from collections.abc import Callable
|
|
||||||
from dataclasses import dataclass
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
from clan_cli.api import API
|
|
||||||
from clan_cli.dirs import TemplateType, clan_templates
|
|
||||||
from clan_cli.errors import ClanError
|
|
||||||
from clan_cli.machines.hardware import HardwareConfig, show_machine_hardware_config
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
# @API.register
|
||||||
class Placeholder:
|
# def set_single_disk_uuid(
|
||||||
# Input name for the user
|
# base_path: str,
|
||||||
label: str
|
# machine_name: str,
|
||||||
options: list[str] | None
|
# disk_uuid: str,
|
||||||
required: bool
|
# ) -> None:
|
||||||
|
# """
|
||||||
|
# Set the disk UUID of single disk machine
|
||||||
|
# """
|
||||||
|
# inventory = load_inventory_json(base_path)
|
||||||
|
|
||||||
|
# instance_name = get_instance_name(machine_name)
|
||||||
|
|
||||||
@dataclass
|
# single_disk_config: ServiceSingleDisk = ServiceSingleDisk(
|
||||||
class DiskSchema:
|
# meta=ServiceMeta(name=instance_name),
|
||||||
name: str
|
# roles=ServiceSingleDiskRole(
|
||||||
placeholders: dict[str, Placeholder]
|
# default=ServiceSingleDiskRoleDefault(
|
||||||
|
# config=SingleDiskConfig(device=f"/dev/disk/by-id/{disk_uuid}"),
|
||||||
|
# machines=[machine_name],
|
||||||
|
# )
|
||||||
|
# ),
|
||||||
|
# )
|
||||||
|
|
||||||
|
# inventory.services.single_disk[instance_name] = single_disk_config
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
# save_inventory(
|
||||||
|
# inventory,
|
||||||
|
# base_path,
|
||||||
def disk_in_facter_report(hw_report: dict) -> bool:
|
# f"Set disk UUID: '{disk_uuid}' on machine: '{machine_name}'",
|
||||||
return "hardware" in hw_report and "disk" in hw_report["hardware"]
|
# )
|
||||||
|
|
||||||
|
|
||||||
def get_best_unix_device_name(unix_device_names: list[str]) -> str:
|
|
||||||
# find the first device name that is disk/by-id
|
|
||||||
for device_name in unix_device_names:
|
|
||||||
if "disk/by-id" in device_name:
|
|
||||||
return device_name
|
|
||||||
else:
|
|
||||||
# if no by-id found, use the first device name
|
|
||||||
return unix_device_names[0]
|
|
||||||
|
|
||||||
|
|
||||||
def hw_main_disk_options(hw_report: dict) -> list[str]:
|
|
||||||
if not disk_in_facter_report(hw_report):
|
|
||||||
msg = "hw_report doesnt include 'disk' information"
|
|
||||||
raise ClanError(msg, description=f"{hw_report.keys()}")
|
|
||||||
|
|
||||||
disks = hw_report["hardware"]["disk"]
|
|
||||||
|
|
||||||
options: list[str] = []
|
|
||||||
for disk in disks:
|
|
||||||
unix_device_names = disk["unix_device_names"]
|
|
||||||
device_name = get_best_unix_device_name(unix_device_names)
|
|
||||||
options += [device_name]
|
|
||||||
|
|
||||||
return options
|
|
||||||
|
|
||||||
|
|
||||||
# must be manually kept in sync with the ${clancore}/templates/disks directory
|
|
||||||
templates: dict[str, dict[str, Callable[[dict[str, Any]], Placeholder]]] = {
|
|
||||||
"single-disk": {
|
|
||||||
# Placeholders
|
|
||||||
"mainDisk": lambda hw_report: Placeholder(
|
|
||||||
label="Main disk", options=hw_main_disk_options(hw_report), required=True
|
|
||||||
),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@API.register
|
|
||||||
def get_disk_schemas(base_path: Path, machine_name: str) -> dict[str, DiskSchema]:
|
|
||||||
"""
|
|
||||||
Get the available disk schemas
|
|
||||||
"""
|
|
||||||
disk_templates = clan_templates(TemplateType.DISK)
|
|
||||||
disk_schemas = {}
|
|
||||||
|
|
||||||
hw_report_path = HardwareConfig.NIXOS_FACTER.config_path(base_path, machine_name)
|
|
||||||
if not hw_report_path.exists():
|
|
||||||
msg = "Hardware configuration missing"
|
|
||||||
raise ClanError(msg)
|
|
||||||
|
|
||||||
hw_report = {}
|
|
||||||
with hw_report_path.open("r") as hw_report_file:
|
|
||||||
hw_report = json.load(hw_report_file)
|
|
||||||
|
|
||||||
for disk_template in disk_templates.iterdir():
|
|
||||||
if disk_template.is_file():
|
|
||||||
schema_name = disk_template.stem
|
|
||||||
if schema_name not in templates:
|
|
||||||
msg = f"Disk schema {schema_name} not found in templates"
|
|
||||||
raise ClanError(
|
|
||||||
msg,
|
|
||||||
description="This is an internal architecture problem. Because disk schemas dont define their own interface",
|
|
||||||
)
|
|
||||||
|
|
||||||
placeholder_getters = templates.get(schema_name)
|
|
||||||
placeholders = {}
|
|
||||||
|
|
||||||
if placeholder_getters:
|
|
||||||
placeholders = {k: v(hw_report) for k, v in placeholder_getters.items()}
|
|
||||||
|
|
||||||
disk_schemas[schema_name] = DiskSchema(
|
|
||||||
name=schema_name, placeholders=placeholders
|
|
||||||
)
|
|
||||||
|
|
||||||
return disk_schemas
|
|
||||||
|
|
||||||
|
|
||||||
@API.register
|
|
||||||
def set_machine_disk_schema(
|
|
||||||
base_path: Path,
|
|
||||||
machine_name: str,
|
|
||||||
schema_name: str,
|
|
||||||
# Placeholders are used to fill in the disk schema
|
|
||||||
placeholders: dict[str, str],
|
|
||||||
) -> None:
|
|
||||||
"""
|
|
||||||
Set the disk placeholders of the template
|
|
||||||
"""
|
|
||||||
# Assert the hw-config must exist before setting the disk
|
|
||||||
hw_config = show_machine_hardware_config(base_path, machine_name)
|
|
||||||
hw_config_path = hw_config.config_path(base_path, machine_name)
|
|
||||||
|
|
||||||
if not hw_config_path.exists():
|
|
||||||
msg = "Hardware configuration must exist before applying disk schema"
|
|
||||||
raise ClanError(msg)
|
|
||||||
|
|
||||||
if hw_config != HardwareConfig.NIXOS_FACTER:
|
|
||||||
msg = "Hardware configuration must use type FACTER for applying disk schema automatically"
|
|
||||||
raise ClanError(msg)
|
|
||||||
|
|
||||||
disk_schema_path = clan_templates(TemplateType.DISK) / f"{schema_name}.nix"
|
|
||||||
|
|
||||||
if not disk_schema_path.exists():
|
|
||||||
msg = f"Disk schema not found at {disk_schema_path}"
|
|
||||||
raise ClanError(msg)
|
|
||||||
|
|
||||||
with disk_schema_path.open("r") as disk_template:
|
|
||||||
print(disk_template.read())
|
|
||||||
|
|||||||
@@ -48,7 +48,6 @@ def find_toplevel(top_level_files: list[str]) -> Path | None:
|
|||||||
|
|
||||||
class TemplateType(Enum):
|
class TemplateType(Enum):
|
||||||
CLAN = "clan"
|
CLAN = "clan"
|
||||||
DISK = "disk"
|
|
||||||
|
|
||||||
|
|
||||||
def clan_templates(template_type: TemplateType) -> Path:
|
def clan_templates(template_type: TemplateType) -> Path:
|
||||||
|
|||||||
@@ -25,6 +25,12 @@
|
|||||||
mountpoint = "/";
|
mountpoint = "/";
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
swap = {
|
||||||
|
size = "{{swapSize}}";
|
||||||
|
content = {
|
||||||
|
type = "swap";
|
||||||
|
};
|
||||||
|
};
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user