cli: add command to list state

Add a subcommand to list configured state for a specific machine.

Example:
```
$ clan state list [MACHINE]
```
This commit is contained in:
a-kenji 2024-06-19 22:31:49 +02:00 committed by kenji
parent 4983c6d302
commit 3bcaeda737
5 changed files with 196 additions and 1 deletions

View File

@ -74,17 +74,18 @@ nav:
- reference/clanModules/zerotier-static-peers.md
- reference/clanModules/zt-tcp-relay.md
- CLI:
- reference/cli/index.md
- reference/cli/backups.md
- reference/cli/config.md
- reference/cli/facts.md
- reference/cli/flakes.md
- reference/cli/flash.md
- reference/cli/history.md
- reference/cli/index.md
- reference/cli/machines.md
- reference/cli/secrets.md
- reference/cli/show.md
- reference/cli/ssh.md
- reference/cli/state.md
- reference/cli/vms.md
- Clan Core:
- reference/clan-core/index.md

View File

@ -21,6 +21,7 @@ from . import (
history,
machines,
secrets,
state,
vms,
)
from .custom_logger import setup_logging
@ -307,6 +308,38 @@ For more detailed information, visit: https://docs.clan.lol/getting-started/depl
)
flash.register_parser(parser_flash)
parser_state = subparsers.add_parser(
"state",
help="query state information about machines",
description="query state information about machines",
epilog=(
"""
This subcommand provides an interface to the state managed by clan.
State can be folders and databases that modules depend on managed by clan.
State directories can be added to on a per machine basis:
```
config.clanCore.state.[SERVICE_NAME].folders = [
"/home"
"/root"
];
```
Here [SERVICE_NAME] can be set freely, if the user sets them extra `userdata`
can be a good choice.
Examples:
$ clan state list [MACHINE]
List state of the machines managed by clan.
For more detailed information, visit: https://docs.clan.lol/getting-started/backups
"""
),
formatter_class=argparse.RawTextHelpFormatter,
)
state.register_parser(parser_state)
if argcomplete:
argcomplete.autocomplete(parser)

View File

@ -160,6 +160,48 @@ def complete_backup_providers_for_machine(
return providers_dict
def complete_state_services_for_machine(
prefix: str, parsed_args: argparse.Namespace, **kwargs: Any
) -> Iterable[str]:
"""
Provides completion functionality for machine state providers.
"""
providers: list[str] = []
machine: str = parsed_args.machine
def run_cmd() -> None:
try:
if (clan_dir_result := clan_dir(None)) is not None:
flake = clan_dir_result
else:
flake = "."
providers_result = json.loads(
run(
nix_eval(
flags=[
f"{flake}#nixosConfigurations.{machine}.config.clan.core.state",
"--apply",
"builtins.attrNames",
],
),
).stdout.strip()
)
providers.extend(providers_result)
except subprocess.CalledProcessError:
pass
thread = threading.Thread(target=run_cmd)
thread.start()
thread.join(timeout=COMPLETION_TIMEOUT)
if thread.is_alive():
return iter([])
providers_dict = {name: "service" for name in providers}
return providers_dict
def complete_secrets(
prefix: str, parsed_args: argparse.Namespace, **kwargs: Any
) -> Iterable[str]:

View File

@ -0,0 +1,34 @@
# !/usr/bin/env python3
import argparse
from .list import register_state_parser
def register_parser(parser: argparse.ArgumentParser) -> None:
subparser = parser.add_subparsers(
title="command",
description="the command to run",
help="the command to run",
required=True,
)
state_parser = subparser.add_parser(
"list",
help="list state folders and the services that configure them",
description="list state folders and the services that configure them",
epilog=(
"""
List state of the machines managed by clan.
Examples:
$ clan state list [MACHINE]
List state of the machine [MACHINE] managed by clan.
For more detailed information, visit: https://docs.clan.lol/getting-started/backups/
"""
),
formatter_class=argparse.RawTextHelpFormatter,
)
register_state_parser(state_parser)

View File

@ -0,0 +1,85 @@
import argparse
import json
import logging
from pathlib import Path
from ..cmd import run_no_stdout
from ..completions import (
add_dynamic_completer,
complete_machines,
complete_state_services_for_machine,
)
from ..dirs import get_clan_flake_toplevel_or_env
from ..errors import ClanCmdError, ClanError
from ..nix import nix_eval
log = logging.getLogger(__name__)
def list_state_folders(machine: str, service: None | str = None) -> None:
uri = "TODO"
if (clan_dir_result := get_clan_flake_toplevel_or_env()) is not None:
flake = clan_dir_result
else:
flake = Path(".")
cmd = nix_eval(
[
f"{flake}#nixosConfigurations.{machine}.config.clanCore.state",
"--json",
]
)
res = "{}"
try:
proc = run_no_stdout(cmd)
res = proc.stdout.strip()
except ClanCmdError:
raise ClanError(
"Clan might not have meta attributes",
location=f"show_clan {uri}",
description="Evaluation failed on clanInternals.meta attribute",
)
state = json.loads(res)
if service:
if state_info := state.get(service):
state = {service: state_info}
else:
raise ClanError(
f"Service {service} isn't configured for this machine.",
location=f"clan state list {machine} --service {service}",
description=f"The service: {service} needs to be configured for the machine.",
)
for service in state:
print(f"· service: {service}")
if folders := state.get(service)["folders"]:
print(" folders:")
for folder in folders:
print(f" - {folder}")
if pre_backup := state.get(service)["preBackupCommand"]:
print(f"preBackupCommand: {pre_backup}")
if pre_restore := state.get(service)["preRestoreCommand"]:
print(f"preRestoreCommand: {pre_restore}")
if post_restore := state.get(service)["postRestoreCommand"]:
print(f"postRestoreCommand: {post_restore}")
print("")
def list_command(args: argparse.Namespace) -> None:
list_state_folders(machine=args.machine, service=args.service)
def register_state_parser(parser: argparse.ArgumentParser) -> None:
machines_parser = parser.add_argument(
"machine",
help="The machine to list state files for",
)
add_dynamic_completer(machines_parser, complete_machines)
service_parser = parser.add_argument(
"--service",
help="the service to show state files for",
)
add_dynamic_completer(service_parser, complete_state_services_for_machine)
parser.set_defaults(func=list_command)