Clan-api: init set disk_schema from facter report

machines/test-inventory-machine/facter.json

@@ -1,4 +1,28 @@
 {
   "version": 1,
-  "system": "x86_64-linux"
+  "system": "x86_64-linux",
+  "virtualisation": "none",
+  "hardware": {
+    "disk": [
+      {
+        "bus_type": {
+          "name": "NVME",
+          "value": 150
+        },
+        "device": {
+          "name": "SAMSUNG MZVL21T0HCLR-00BL7",
+          "value": 43018
+        },
+        "model": "SAMSUNG MZVL21T0HCLR-00BL7",
+        "unix_device_names": [
+          "/dev/disk/by-diskseq/1",
+          "/dev/disk/by-id/nvme-SAMSUNG_MZVL21T0HCLR-00BL7",
+          "/dev/disk/by-id/nvme-SAMSUNG_MZVL21T0HCLR-00BL7",
+          "/dev/disk/by-id/nvme-eui.0087324",
+          "/dev/disk/by-path/pci-0000:02:00.0-nvme-1",
+          "/dev/nvme0n1"
+        ]
+      }
+    ]
+  }
 }

pkgs/clan-cli/clan_cli/api/disk.py

@@ -1,34 +1,192 @@
-def get_instance_name(machine_name: str) -> str:
+import json
-    return f"{machine_name}-single-disk"
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from clan_cli.api import API
+from clan_cli.dirs import TemplateType, clan_templates
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_file
+from clan_cli.machines.hardware import HardwareConfig, show_machine_hardware_config
+
+log = logging.getLogger(__name__)
 
 
-# @API.register
+def disk_in_facter_report(hw_report: dict) -> bool:
-# def set_single_disk_uuid(
+    return "hardware" in hw_report and "disk" in hw_report["hardware"]
-#     base_path: str,
-#     machine_name: str,
-#     disk_uuid: str,
-# ) -> None:
-#     """
-#     Set the disk UUID of single disk machine
-#     """
-#     inventory = load_inventory_json(base_path)
 
-#     instance_name = get_instance_name(machine_name)
 
-#     single_disk_config: ServiceSingleDisk = ServiceSingleDisk(
+def get_best_unix_device_name(unix_device_names: list[str]) -> str:
-#         meta=ServiceMeta(name=instance_name),
+    # find the first device name that is disk/by-id
-#         roles=ServiceSingleDiskRole(
+    for device_name in unix_device_names:
-#             default=ServiceSingleDiskRoleDefault(
+        if "disk/by-id" in device_name:
-#                 config=SingleDiskConfig(device=f"/dev/disk/by-id/{disk_uuid}"),
+            return device_name
-#                 machines=[machine_name],
+    else:
-#             )
+        # if no by-id found, use the first device name
-#         ),
+        return unix_device_names[0]
-#     )
 
-#     inventory.services.single_disk[instance_name] = single_disk_config
 
-#     save_inventory(
+def hw_main_disk_options(hw_report: dict) -> list[str]:
-#         inventory,
+    if not disk_in_facter_report(hw_report):
-#         base_path,
+        msg = "hw_report doesnt include 'disk' information"
-#         f"Set disk UUID: '{disk_uuid}' on machine: '{machine_name}'",
+        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
+
+
+@dataclass
+class Placeholder:
+    # Input name for the user
+    label: str
+    options: list[str] | None
+    required: bool
+
+
+@dataclass
+class DiskSchema:
+    name: str
+    placeholders: dict[str, Placeholder]
+
+
+# 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
+    # Use get disk schemas to get the placeholders and their options
+    placeholders: dict[str, str],
+    force: bool = False,
+) -> 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)
+
+    # Check that the placeholders are valid
+    disk_schema = get_disk_schemas(base_path, machine_name)[schema_name]
+    # check that all required placeholders are present
+    for placeholder_name, schema_placeholder in disk_schema.placeholders.items():
+        if schema_placeholder.required and placeholder_name not in placeholders:
+            msg = f"Required placeholder {placeholder_name} - {schema_placeholder} missing"
+            raise ClanError(msg)
+
+    # For every placeholder check that the value is valid
+    for placeholder_name, placeholder_value in placeholders.items():
+        ph = disk_schema.placeholders.get(placeholder_name)
+        # Unknown placeholder
+        if not ph:
+            msg = (
+                f"Placeholder {placeholder_name} not found in disk schema {schema_name}"
+            )
+            raise ClanError(msg)
+
+        # Invalid value. Check if the value is one of the provided options
+        if ph.options and placeholder_value not in ph.options:
+            msg = (
+                f"Invalid value {placeholder_value} for placeholder {placeholder_name}"
+            )
+            raise ClanError(msg, description=f"Valid options: {ph.options}")
+
+    with disk_schema_path.open("r") as disk_template:
+        config_str = disk_template.read()
+        for placeholder_name, placeholder_value in placeholders.items():
+            config_str = config_str.replace(
+                r"{{" + placeholder_name + r"}}", placeholder_value
+            )
+
+        # Custom replacements
+        config_str = config_str.replace(r"{{uuid}}", str(uuid4()).replace("-", ""))
+
+        # place disko.nix alongside the hw-config
+        disko_file_path = hw_config_path.parent.joinpath("disko.nix")
+        if disko_file_path.exists() and not force:
+            msg = f"Disk schema already exists at {disko_file_path}"
+            raise ClanError(msg)
+
+        with disko_file_path.open("w") as disk_config:
+            disk_config.write(config_str)
+
+        commit_file(
+            disko_file_path,
+            base_path,
+            commit_message=f"Set disk schema of machine: {machine_name} to {schema_name}",
+        )

pkgs/clan-cli/clan_cli/dirs.py

@@ -48,6 +48,7 @@ def find_toplevel(top_level_files: list[str]) -> Path | None:
 
 class TemplateType(Enum):
     CLAN = "clan"
+    DISK = "disk"
 
 
 def clan_templates(template_type: TemplateType) -> Path:

templates/disk/single-disk.nix

@@ -2,6 +2,7 @@
   disko.devices = {
     disk = {
       main = {
+        name = "main-{{uuid}}";
         device = "{{mainDisk}}";
         type = "disk";
         content = {
@@ -25,12 +26,6 @@
                 mountpoint = "/";
               };
             };
-            swap = {
-              size = "{{swapSize}}";
-              content = {
-                type = "swap";
-              };
-            };
           };
         };
       };