Merge pull request 'Move list.py to clan_lib/machines part 2' (#4068) from Qubasa/clan-core:move_to_clan_lib4 into main
Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/4068
This commit is contained in:
Luis Hebendanz
@@ -1,113 +1,14 @@
|
||||
import argparse
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from clan_lib.api import API
|
||||
from clan_lib.api.disk import MachineDiskMatter
|
||||
from clan_lib.api.modules import parse_frontmatter
|
||||
from clan_lib.dirs import specific_machine_dir
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.actions import get_machine, list_machines
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.nix_models.clan import InventoryMachine
|
||||
from clan_lib.machines.list import list_full_machines, query_machines_by_tags
|
||||
|
||||
from clan_cli.completions import add_dynamic_completer, complete_tags
|
||||
from clan_cli.machines.hardware import HardwareConfig
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def list_full_machines(
|
||||
flake: Flake, nix_options: list[str] | None = None
|
||||
) -> dict[str, Machine]:
|
||||
"""
|
||||
Like `list_machines`, but returns a full 'machine' instance for each machine.
|
||||
"""
|
||||
machines = list_machines(flake)
|
||||
|
||||
res: dict[str, Machine] = {}
|
||||
|
||||
if nix_options is None:
|
||||
nix_options = []
|
||||
|
||||
for inv_machine in machines.values():
|
||||
name = inv_machine.get("name")
|
||||
# Technically, this should not happen, but we are defensive here.
|
||||
if name is None:
|
||||
msg = "InternalError: Machine name is required. But got a machine without a name."
|
||||
raise ClanError(msg)
|
||||
|
||||
machine = Machine(
|
||||
name=name,
|
||||
flake=flake,
|
||||
nix_options=nix_options,
|
||||
)
|
||||
res[machine.name] = machine
|
||||
|
||||
return res
|
||||
|
||||
|
||||
def query_machines_by_tags(flake: Flake, tags: list[str]) -> dict[str, Machine]:
|
||||
"""
|
||||
Query machines by their respective tags, if multiple tags are specified
|
||||
then only machines that have those respective tags specified will be listed.
|
||||
It is an intersection of the tags and machines.
|
||||
"""
|
||||
machines = list_full_machines(flake)
|
||||
|
||||
filtered_machines = {}
|
||||
for machine in machines.values():
|
||||
inv_machine = get_machine(machine.flake, machine.name)
|
||||
machine_tags = inv_machine.get("tags", [])
|
||||
if all(tag in machine_tags for tag in tags):
|
||||
filtered_machines[machine.name] = machine
|
||||
|
||||
return filtered_machines
|
||||
|
||||
|
||||
@dataclass
|
||||
class MachineDetails:
|
||||
machine: InventoryMachine
|
||||
hw_config: HardwareConfig | None = None
|
||||
disk_schema: MachineDiskMatter | None = None
|
||||
|
||||
|
||||
def extract_header(c: str) -> str:
|
||||
header_lines = []
|
||||
for line in c.splitlines():
|
||||
match = re.match(r"^\s*#(.*)", line)
|
||||
if match:
|
||||
header_lines.append(match.group(1).strip())
|
||||
else:
|
||||
break # Stop once the header ends
|
||||
return "\n".join(header_lines)
|
||||
|
||||
|
||||
@API.register
|
||||
def get_machine_details(machine: Machine) -> MachineDetails:
|
||||
machine_inv = get_machine(machine.flake, machine.name)
|
||||
hw_config = HardwareConfig.detect_type(machine)
|
||||
|
||||
machine_dir = specific_machine_dir(machine)
|
||||
disk_schema: MachineDiskMatter | None = None
|
||||
disk_path = machine_dir / "disko.nix"
|
||||
if disk_path.exists():
|
||||
with disk_path.open() as f:
|
||||
content = f.read()
|
||||
header = extract_header(content)
|
||||
data, _rest = parse_frontmatter(header)
|
||||
if data:
|
||||
disk_schema = data # type: ignore
|
||||
|
||||
return MachineDetails(
|
||||
machine=machine_inv,
|
||||
hw_config=hw_config,
|
||||
disk_schema=disk_schema,
|
||||
)
|
||||
|
||||
|
||||
def list_command(args: argparse.Namespace) -> None:
|
||||
flake: Flake = args.flake
|
||||
|
||||
|
||||
@@ -1,140 +1,14 @@
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from clan_lib.cmd import Log, RunOpts, run
|
||||
from clan_lib.dirs import get_clan_flake_toplevel_or_env, specific_machine_dir
|
||||
from clan_lib.dirs import get_clan_flake_toplevel_or_env
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.actions import list_machines
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.nix import nix_build, nix_command
|
||||
from clan_lib.nix_models.clan import InventoryMachine
|
||||
|
||||
from clan_cli.machines.create import CreateOptions, create_machine
|
||||
from clan_cli.vars.generate import generate_vars
|
||||
from clan_lib.machines.morph import morph_machine
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def is_local_input(node: dict[str, dict[str, str]]) -> bool:
|
||||
locked = node.get("locked")
|
||||
if not locked:
|
||||
return False
|
||||
# matches path and git+file://
|
||||
return (
|
||||
locked["type"] == "path"
|
||||
or re.match(r"^\w+\+file://", locked.get("url", "")) is not None
|
||||
)
|
||||
|
||||
|
||||
def random_hostname() -> str:
|
||||
adjectives = ["wacky", "happy", "fluffy", "silly", "quirky", "zany", "bouncy"]
|
||||
nouns = ["unicorn", "penguin", "goose", "ninja", "octopus", "hamster", "robot"]
|
||||
adjective = random.choice(adjectives)
|
||||
noun = random.choice(nouns)
|
||||
return f"{adjective}-{noun}"
|
||||
|
||||
|
||||
def morph_machine(
|
||||
flake: Flake, template: str, ask_confirmation: bool, name: str | None = None
|
||||
) -> None:
|
||||
cmd = nix_command(
|
||||
[
|
||||
"flake",
|
||||
"archive",
|
||||
"--json",
|
||||
f"{flake}",
|
||||
]
|
||||
)
|
||||
|
||||
archive_json = run(
|
||||
cmd, RunOpts(error_msg="Failed to archive flake for morphing")
|
||||
).stdout.rstrip()
|
||||
archive_path = json.loads(archive_json)["path"]
|
||||
|
||||
with TemporaryDirectory(prefix="morph-") as _temp_dir:
|
||||
flakedir = Path(_temp_dir).resolve() / "flake"
|
||||
|
||||
flakedir.mkdir(parents=True, exist_ok=True)
|
||||
run(["cp", "-r", archive_path + "/.", str(flakedir)])
|
||||
run(["chmod", "-R", "+w", str(flakedir)])
|
||||
|
||||
os.chdir(flakedir)
|
||||
|
||||
if name is None:
|
||||
name = random_hostname()
|
||||
|
||||
if name not in list_machines(flake):
|
||||
create_opts = CreateOptions(
|
||||
template=template,
|
||||
machine=InventoryMachine(name=name),
|
||||
clan_dir=Flake(str(flakedir)),
|
||||
)
|
||||
create_machine(create_opts, commit=False)
|
||||
|
||||
machine = Machine(name=name, flake=Flake(str(flakedir)))
|
||||
|
||||
generate_vars([machine], generator_name=None, regenerate=False)
|
||||
|
||||
machine.secret_vars_store.populate_dir(
|
||||
output_dir=Path("/run/secrets"), phases=["activation", "users", "services"]
|
||||
)
|
||||
|
||||
# run(["nixos-facter", "-o", f"{flakedir}/machines/{name}/facter.json"]).stdout
|
||||
|
||||
# facter_json = run(["nixos-facter"]).stdout
|
||||
# run(["cp", "facter.json", f"{flakedir}/machines/{name}/facter.json"]).stdout
|
||||
|
||||
machine_dir = specific_machine_dir(machine)
|
||||
machine_dir.mkdir(parents=True, exist_ok=True)
|
||||
Path(f"{machine_dir}/facter.json").write_text('{"system": "x86_64-linux"}')
|
||||
result_path = run(
|
||||
nix_build(
|
||||
[f"{flakedir}#nixosConfigurations.{name}.config.system.build.toplevel"]
|
||||
)
|
||||
).stdout.rstrip()
|
||||
|
||||
ropts = RunOpts(log=Log.BOTH)
|
||||
|
||||
run(
|
||||
[
|
||||
f"{result_path}/sw/bin/nixos-rebuild",
|
||||
"dry-activate",
|
||||
"--flake",
|
||||
f"{flakedir}#{name}",
|
||||
],
|
||||
ropts,
|
||||
).stdout.rstrip()
|
||||
|
||||
if ask_confirmation:
|
||||
log.warning("ARE YOU SURE YOU WANT TO DO THIS?")
|
||||
log.warning(
|
||||
"You should have read and understood all of the above and know what you are doing."
|
||||
)
|
||||
|
||||
ask = input(
|
||||
f"Do you really want convert this machine into {name}? If to continue, type in the new machine name: "
|
||||
)
|
||||
if ask != name:
|
||||
return
|
||||
|
||||
run(
|
||||
[
|
||||
f"{result_path}/sw/bin/nixos-rebuild",
|
||||
"test",
|
||||
"--flake",
|
||||
f"{flakedir}#{name}",
|
||||
],
|
||||
ropts,
|
||||
).stdout.rstrip()
|
||||
|
||||
|
||||
def morph_command(args: argparse.Namespace) -> None:
|
||||
if args.flake:
|
||||
clan_dir = args.flake
|
||||
|
||||
106
pkgs/clan-cli/clan_lib/machines/list.py
Normal file
106
pkgs/clan-cli/clan_lib/machines/list.py
Normal file
@@ -0,0 +1,106 @@
|
||||
import logging
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
|
||||
from clan_cli.machines.hardware import HardwareConfig
|
||||
|
||||
from clan_lib.api import API
|
||||
from clan_lib.api.disk import MachineDiskMatter
|
||||
from clan_lib.api.modules import parse_frontmatter
|
||||
from clan_lib.dirs import specific_machine_dir
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.actions import get_machine, list_machines
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.nix_models.clan import InventoryMachine
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def list_full_machines(
|
||||
flake: Flake, nix_options: list[str] | None = None
|
||||
) -> dict[str, Machine]:
|
||||
"""
|
||||
Like `list_machines`, but returns a full 'machine' instance for each machine.
|
||||
"""
|
||||
machines = list_machines(flake)
|
||||
|
||||
res: dict[str, Machine] = {}
|
||||
|
||||
if nix_options is None:
|
||||
nix_options = []
|
||||
|
||||
for inv_machine in machines.values():
|
||||
name = inv_machine.get("name")
|
||||
# Technically, this should not happen, but we are defensive here.
|
||||
if name is None:
|
||||
msg = "InternalError: Machine name is required. But got a machine without a name."
|
||||
raise ClanError(msg)
|
||||
|
||||
machine = Machine(
|
||||
name=name,
|
||||
flake=flake,
|
||||
nix_options=nix_options,
|
||||
)
|
||||
res[machine.name] = machine
|
||||
|
||||
return res
|
||||
|
||||
|
||||
def query_machines_by_tags(flake: Flake, tags: list[str]) -> dict[str, Machine]:
|
||||
"""
|
||||
Query machines by their respective tags, if multiple tags are specified
|
||||
then only machines that have those respective tags specified will be listed.
|
||||
It is an intersection of the tags and machines.
|
||||
"""
|
||||
machines = list_full_machines(flake)
|
||||
|
||||
filtered_machines = {}
|
||||
for machine in machines.values():
|
||||
inv_machine = get_machine(machine.flake, machine.name)
|
||||
machine_tags = inv_machine.get("tags", [])
|
||||
if all(tag in machine_tags for tag in tags):
|
||||
filtered_machines[machine.name] = machine
|
||||
|
||||
return filtered_machines
|
||||
|
||||
|
||||
@dataclass
|
||||
class MachineDetails:
|
||||
machine: InventoryMachine
|
||||
hw_config: HardwareConfig | None = None
|
||||
disk_schema: MachineDiskMatter | None = None
|
||||
|
||||
|
||||
def extract_header(c: str) -> str:
|
||||
header_lines = []
|
||||
for line in c.splitlines():
|
||||
match = re.match(r"^\s*#(.*)", line)
|
||||
if match:
|
||||
header_lines.append(match.group(1).strip())
|
||||
else:
|
||||
break # Stop once the header ends
|
||||
return "\n".join(header_lines)
|
||||
|
||||
|
||||
@API.register
|
||||
def get_machine_details(machine: Machine) -> MachineDetails:
|
||||
machine_inv = get_machine(machine.flake, machine.name)
|
||||
hw_config = HardwareConfig.detect_type(machine)
|
||||
|
||||
machine_dir = specific_machine_dir(machine)
|
||||
disk_schema: MachineDiskMatter | None = None
|
||||
disk_path = machine_dir / "disko.nix"
|
||||
if disk_path.exists():
|
||||
with disk_path.open() as f:
|
||||
content = f.read()
|
||||
header = extract_header(content)
|
||||
data, _rest = parse_frontmatter(header)
|
||||
if data:
|
||||
disk_schema = data # type: ignore
|
||||
|
||||
return MachineDetails(
|
||||
machine=machine_inv,
|
||||
hw_config=hw_config,
|
||||
disk_schema=disk_schema,
|
||||
)
|
||||
133
pkgs/clan-cli/clan_lib/machines/morph.py
Normal file
133
pkgs/clan-cli/clan_lib/machines/morph.py
Normal file
@@ -0,0 +1,133 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
from clan_cli.machines.create import CreateOptions, create_machine
|
||||
from clan_cli.vars.generate import generate_vars
|
||||
|
||||
from clan_lib.cmd import Log, RunOpts, run
|
||||
from clan_lib.dirs import specific_machine_dir
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.machines.actions import list_machines
|
||||
from clan_lib.machines.machines import Machine
|
||||
from clan_lib.nix import nix_build, nix_command
|
||||
from clan_lib.nix_models.clan import InventoryMachine
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def is_local_input(node: dict[str, dict[str, str]]) -> bool:
|
||||
locked = node.get("locked")
|
||||
if not locked:
|
||||
return False
|
||||
# matches path and git+file://
|
||||
return (
|
||||
locked["type"] == "path"
|
||||
or re.match(r"^\w+\+file://", locked.get("url", "")) is not None
|
||||
)
|
||||
|
||||
|
||||
def random_hostname() -> str:
|
||||
adjectives = ["wacky", "happy", "fluffy", "silly", "quirky", "zany", "bouncy"]
|
||||
nouns = ["unicorn", "penguin", "goose", "ninja", "octopus", "hamster", "robot"]
|
||||
adjective = random.choice(adjectives)
|
||||
noun = random.choice(nouns)
|
||||
return f"{adjective}-{noun}"
|
||||
|
||||
|
||||
def morph_machine(
|
||||
flake: Flake, template: str, ask_confirmation: bool, name: str | None = None
|
||||
) -> None:
|
||||
cmd = nix_command(
|
||||
[
|
||||
"flake",
|
||||
"archive",
|
||||
"--json",
|
||||
f"{flake}",
|
||||
]
|
||||
)
|
||||
|
||||
archive_json = run(
|
||||
cmd, RunOpts(error_msg="Failed to archive flake for morphing")
|
||||
).stdout.rstrip()
|
||||
archive_path = json.loads(archive_json)["path"]
|
||||
|
||||
with TemporaryDirectory(prefix="morph-") as _temp_dir:
|
||||
flakedir = Path(_temp_dir).resolve() / "flake"
|
||||
|
||||
flakedir.mkdir(parents=True, exist_ok=True)
|
||||
run(["cp", "-r", archive_path + "/.", str(flakedir)])
|
||||
run(["chmod", "-R", "+w", str(flakedir)])
|
||||
|
||||
os.chdir(flakedir)
|
||||
|
||||
if name is None:
|
||||
name = random_hostname()
|
||||
|
||||
if name not in list_machines(flake):
|
||||
create_opts = CreateOptions(
|
||||
template=template,
|
||||
machine=InventoryMachine(name=name),
|
||||
clan_dir=Flake(str(flakedir)),
|
||||
)
|
||||
create_machine(create_opts, commit=False)
|
||||
|
||||
machine = Machine(name=name, flake=Flake(str(flakedir)))
|
||||
|
||||
generate_vars([machine], generator_name=None, regenerate=False)
|
||||
|
||||
machine.secret_vars_store.populate_dir(
|
||||
output_dir=Path("/run/secrets"), phases=["activation", "users", "services"]
|
||||
)
|
||||
|
||||
# run(["nixos-facter", "-o", f"{flakedir}/machines/{name}/facter.json"]).stdout
|
||||
|
||||
# facter_json = run(["nixos-facter"]).stdout
|
||||
# run(["cp", "facter.json", f"{flakedir}/machines/{name}/facter.json"]).stdout
|
||||
|
||||
machine_dir = specific_machine_dir(machine)
|
||||
machine_dir.mkdir(parents=True, exist_ok=True)
|
||||
Path(f"{machine_dir}/facter.json").write_text('{"system": "x86_64-linux"}')
|
||||
result_path = run(
|
||||
nix_build(
|
||||
[f"{flakedir}#nixosConfigurations.{name}.config.system.build.toplevel"]
|
||||
)
|
||||
).stdout.rstrip()
|
||||
|
||||
ropts = RunOpts(log=Log.BOTH)
|
||||
|
||||
run(
|
||||
[
|
||||
f"{result_path}/sw/bin/nixos-rebuild",
|
||||
"dry-activate",
|
||||
"--flake",
|
||||
f"{flakedir}#{name}",
|
||||
],
|
||||
ropts,
|
||||
).stdout.rstrip()
|
||||
|
||||
if ask_confirmation:
|
||||
log.warning("ARE YOU SURE YOU WANT TO DO THIS?")
|
||||
log.warning(
|
||||
"You should have read and understood all of the above and know what you are doing."
|
||||
)
|
||||
|
||||
ask = input(
|
||||
f"Do you really want convert this machine into {name}? If to continue, type in the new machine name: "
|
||||
)
|
||||
if ask != name:
|
||||
return
|
||||
|
||||
run(
|
||||
[
|
||||
f"{result_path}/sw/bin/nixos-rebuild",
|
||||
"test",
|
||||
"--flake",
|
||||
f"{flakedir}#{name}",
|
||||
],
|
||||
ropts,
|
||||
).stdout.rstrip()
|
||||
@@ -6,11 +6,11 @@ import logging
|
||||
from typing import Any
|
||||
|
||||
from clan_cli.clan.inspect import FlakeConfig, inspect_flake
|
||||
from clan_cli.machines.list import list_machines
|
||||
from clan_lib.dirs import user_history_file
|
||||
from clan_lib.errors import ClanError
|
||||
from clan_lib.flake import Flake
|
||||
from clan_lib.locked_open import read_history_file, write_history_file
|
||||
from clan_lib.machines.list import list_machines
|
||||
|
||||
from clan_vm_manager.clan_uri import ClanURI
|
||||
|
||||
|
||||
Reference in New Issue
Block a user