nixos-rebuild-ng: configure logging

This commit is contained in:
Thiago Kenji Okada
2024-11-30 17:54:29 +00:00
parent 5cc71a346a
commit 2db09d7f77
5 changed files with 53 additions and 26 deletions
@@ -1,6 +1,7 @@
import argparse
import atexit
import json
import logging
import sys
from pathlib import Path
from subprocess import run
@@ -10,9 +11,10 @@ from typing import assert_never
from . import nix
from .models import Action, BuildAttr, Flake, NRError, Profile
from .process import Remote, cleanup_ssh
from .utils import info
from .utils import LogFormatter
VERBOSE = 0
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def get_parser() -> tuple[argparse.ArgumentParser, dict[str, argparse.ArgumentParser]]:
@@ -103,11 +105,11 @@ def parse_args(
}
def parser_warn(msg: str) -> None:
info(f"{parser.prog}: warning: {msg}")
print(f"{parser.prog}: warning: {msg}", file=sys.stderr)
global VERBOSE
# This flag affects both nix and this script
VERBOSE = args.verbose
if args.verbose:
logger.setLevel(logging.DEBUG)
# https://github.com/NixOS/nixpkgs/blob/master/pkgs/os-specific/linux/nixos-rebuild/nixos-rebuild.sh#L56
if args.action == Action.DRY_RUN.value:
@@ -190,7 +192,7 @@ def execute(argv: list[str]) -> None:
match action:
case Action.SWITCH | Action.BOOT:
info("building the system configuration...")
logger.info("building the system configuration...")
if args.rollback:
path_to_config = nix.rollback(profile, target_host, sudo=args.sudo)
else:
@@ -219,7 +221,7 @@ def execute(argv: list[str]) -> None:
install_bootloader=args.install_bootloader,
)
case Action.TEST | Action.BUILD | Action.DRY_BUILD | Action.DRY_ACTIVATE:
info("building the system configuration...")
logger.info("building the system configuration...")
dry_run = action == Action.DRY_BUILD
if args.rollback:
if action not in (Action.TEST, Action.BUILD):
@@ -256,7 +258,7 @@ def execute(argv: list[str]) -> None:
specialisation=args.specialisation,
)
case Action.BUILD_VM | Action.BUILD_VM_WITH_BOOTLOADER:
info("building the system configuration...")
logger.info("building the system configuration...")
attr = "vm" if action == Action.BUILD_VM else "vmWithBootLoader"
if flake:
path_to_config = nix.nixos_build_flake(
@@ -313,10 +315,14 @@ def execute(argv: list[str]) -> None:
def main() -> None:
ch = logging.StreamHandler()
ch.setFormatter(LogFormatter("%(levelname)s: %(message)s"))
logger.addHandler(ch)
try:
execute(sys.argv)
except (Exception, KeyboardInterrupt) as ex:
if VERBOSE:
if logger.level == logging.DEBUG:
raise ex
else:
sys.exit(str(ex))
@@ -1,3 +1,4 @@
import logging
import os
from datetime import datetime
from importlib.resources import files
@@ -17,10 +18,11 @@ from .models import (
Remote,
)
from .process import run_wrapper
from .utils import Args, dict_to_flags, info
from .utils import Args, dict_to_flags
FLAKE_FLAGS: Final = ["--extra-experimental-features", "nix-command flakes"]
FLAKE_REPL_TEMPLATE: Final = "repl.template.nix"
logger = logging.getLogger(__name__)
def copy_closure(
@@ -107,7 +109,7 @@ def get_nixpkgs_rev(nixpkgs_path: Path | None) -> str | None:
)
except FileNotFoundError:
# Git is not included in the closure so we need to check
info(f"warning: Git not found; cannot figure out revision of '{nixpkgs_path}'")
logger.warning(f"Git not found; cannot figure out revision of '{nixpkgs_path}'")
return None
if rev := r.stdout.strip():
@@ -203,13 +205,15 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
)
try:
nixos_version = (generation_path / "nixos-version").read_text().strip()
except IOError:
except IOError as ex:
logger.debug("could not get nixos-version: %s", ex)
nixos_version = "Unknown"
try:
kernel_version = next(
(generation_path / "kernel-modules/lib/modules").iterdir()
).name
except IOError:
except IOError as ex:
logger.debug("could not get kernel version: %s", ex)
kernel_version = "Unknown"
specialisations = [
s.name for s in (generation_path / "specialisation").glob("*") if s.is_dir()
@@ -219,7 +223,8 @@ def list_generations(profile: Profile) -> list[GenerationJson]:
[generation_path / "sw/bin/nixos-version", "--configuration-revision"],
capture_output=True,
).stdout.strip()
except (CalledProcessError, IOError):
except (CalledProcessError, IOError) as ex:
logger.debug("could not get configuration revision: %s", ex)
configuration_revision = "Unknown"
result.append(
@@ -1,5 +1,6 @@
from __future__ import annotations
import logging
import os
import subprocess
from dataclasses import dataclass
@@ -7,7 +8,7 @@ from getpass import getpass
from pathlib import Path
from typing import Self, Sequence, TypedDict, Unpack
from .utils import info
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
@@ -46,13 +47,13 @@ class Remote:
def _validate_opts(opts: list[str], ask_sudo_password: bool | None) -> None:
for o in opts:
if o in ["-t", "-tt", "RequestTTY=yes", "RequestTTY=force"]:
info(
f"warning: detected option '{o}' in NIX_SSHOPTS. SSH's TTY "
+ "may cause issues, it is recommended to remove this option"
logger.warning(
f"detected option '{o}' in NIX_SSHOPTS. SSH's TTY may "
+ "cause issues, it is recommended to remove this option"
)
if not ask_sudo_password:
info(
"If you want to prompt for sudo password use "
logger.warning(
"if you want to prompt for sudo password use "
+ "'--ask-sudo-password' option instead"
)
@@ -99,6 +100,8 @@ def run_wrapper(
if sudo:
args = ["sudo", *args]
logger.debug("calling run with args=%s, extra_env=%s", args, extra_env)
return subprocess.run(
args,
check=check,
@@ -1,11 +1,23 @@
import sys
from functools import partial
from typing import TypeAlias
import logging
from typing import TypeAlias, override
info = partial(print, file=sys.stderr)
Args: TypeAlias = bool | str | list[str] | int | None
class LogFormatter(logging.Formatter):
@override
def format(self, record: logging.LogRecord) -> str:
record.levelname = record.levelname.lower()
match record.levelno:
case logging.INFO:
self._style._fmt = "%(message)s"
case logging.DEBUG:
self._style._fmt = "%(levelname)s: %(name)s: %(message)s"
case _:
self._style._fmt = "%(levelname)s: %(message)s"
return super().format(record)
def dict_to_flags(d: dict[str, Args]) -> list[str]:
flags = []
for key, value in d.items():
@@ -1,3 +1,4 @@
import logging
import textwrap
from pathlib import Path
from subprocess import PIPE, CompletedProcess
@@ -43,7 +44,7 @@ def test_parse_args() -> None:
"bar",
]
)
assert nr.VERBOSE == 0
assert nr.logger.level == logging.INFO
assert r1.flake == "/etc/nixos"
assert r1.install_bootloader is True
assert r1.install_grub is True
@@ -65,7 +66,7 @@ def test_parse_args() -> None:
"-vvv",
]
)
assert nr.VERBOSE == 3
assert nr.logger.level == logging.DEBUG
assert r2.verbose == 3
assert r2.flake is False
assert r2.action == "dry-build"