Richie/dotfiles
1
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Implement ZFS dataset management and snapshot configuration

Browse Source 
- Introduced a new `datasets.nix` file to define ZFS datasets and their properties, allowing for centralized management of dataset configurations.
- Updated `default.nix` to import and enable the `zfs_manager` service, integrating it into the system configuration.
- Refactored `zfs.sh` to remove dataset creation commands, delegating dataset management to the `zfs_manager` service.
- Removed the legacy `snapshot_config.toml` file, as snapshot configurations are now handled within `datasets.nix`.
- Modified `vars.nix` to derive mountpoint paths from `datasets.nix`, ensuring consistency across the configuration.
- Created a new `zfs.nix` file to define the `zfs_manager` service and its dependencies.
- Added comprehensive tests for the `zfs_manager` functionality, covering dataset creation, property management, and error handling.
This commit is contained in:
Richie
2026-08-19 21:29:00 -04:00
parent e5132e2a0b
commit e3602feb5a
10 changed files with 1372 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/tools/snapshot_manager.py
+26 -5
View File
@@ -23,6 +23,7 @@ def main(config_file: Path) -> None:
"""Main."""
configure_logger(level="DEBUG")
logger.info("Starting snapshot_manager")
failures: list[str] = []
try:
time_stamp = get_time_stamp()
@@ -34,16 +35,23 @@ def main(config_file: Path) -> None:
msg = f"{dataset.name} failed to create snapshot {time_stamp}"
logger.error(msg)
signal_alert(msg)
failures.append(msg)
continue
count_lookup = get_count_lookup(config_file, dataset.name)
logger.info(f"using {count_lookup} for {dataset.name}")
get_snapshots_to_delete(dataset, count_lookup)
failures.extend(get_snapshots_to_delete(dataset, count_lookup))
except Exception:
logger.exception("snapshot_manager failed")
signal_alert("snapshot_manager failed")
sys.exit(1)
else:
logger.info("snapshot_manager completed")
if failures:
logger.error(f"snapshot_manager completed with {len(failures)} errors")
for failure in failures:
logger.error(f" {failure}")
sys.exit(1)
logger.info("snapshot_manager completed")
def get_count_lookup(config_file: Path, dataset_name: str) -> dict[str, int]:
@@ -92,19 +100,29 @@ def load_config_data(config_file: Path) -> dict[str, dict[str, int]]:
def get_snapshots_to_delete(
dataset: Dataset,
count_lookup: dict[str, int],
) -> None:
) -> list[str]:
"""Get snapshots to delete.
Args:
dataset (Dataset): the dataset
count_lookup (dict[str, int]): the count lookup
Returns:
list[str]: Snapshot deletion failures encountered while pruning.
"""
for retention_class in ("15_min", "hourly", "daily", "monthly"):
count = count_lookup.get(retention_class)
if not isinstance(count, int) or isinstance(count, bool) or count < 0:
error = f"{retention_class} retention must be a non-negative integer, got {count!r}"
raise ValueError(error)
failures: list[str] = []
snapshots = dataset.get_snapshots()
logger.info(f"calculating snapshots for {dataset.name} to be deleted")
if not snapshots:
logger.info(f"{dataset.name} has no snapshots")
return
return failures
filters = (
("15_min", re_compile(r"auto_\d{10}(?:15|30|45)")),
@@ -129,6 +147,9 @@ def get_snapshots_to_delete(
error_message = f"{dataset.name}@{snapshot} failed to delete: {error}"
signal_alert(error_message)
logger.error(error_message)
failures.append(error_message)
return failures
def get_time_stamp() -> str:
python/tools/zfs_manager.py
+340
View File
@@ -0,0 +1,340 @@
"""zfs_manager.
Reconciles the live zfs datasets against a declaration generated by
common/optional/zfs_manager.nix. Datasets are created and properties are
corrected, but nothing is ever destroyed or renamed.
"""
from __future__ import annotations
import json
import logging
import sys
from pathlib import Path # noqa: TC003 This is required for the typer CLI
import typer
from python.common import configure_logger
from python.signal_alert import signal_alert
from python.zfs import create_dataset, get_properties, list_dataset_names, set_property
logger = logging.getLogger(__name__)
# Properties that can only be chosen at creation time. Attempting to zfs set
# these fails on every run, so a mismatch is reported instead of retried.
CREATE_ONLY_PROPERTIES = frozenset(
{
"casesensitivity",
"encryption",
"keyformat",
"normalization",
"utf8only",
"volblocksize",
},
)
# Properties whose values zfs reports in bytes but which are conventionally
# declared with a size suffix, so "16k" and "16384" mean the same thing.
SIZE_PROPERTIES = frozenset(
{
"quota",
"recordsize",
"refquota",
"refreservation",
"reservation",
"special_small_blocks",
"volblocksize",
"volsize",
},
)
SIZE_SUFFIXES = {"b": 1, "k": 1024, "m": 1024**2, "g": 1024**3, "t": 1024**4, "p": 1024**5}
# Sources that mean the value was deliberately put on this dataset rather than
# inherited from a parent or left at the zfs default.
LOCAL_SOURCES = ("local", "received")
class ReconciliationError(RuntimeError):
"""One or more datasets could not be brought in line with the declaration."""
def __init__(self, failures: list[str]) -> None:
"""Record the individual failures behind this run's exit code."""
self.failures = failures
super().__init__(f"ZFS reconciliation failed with {len(failures)} errors")
def main(config_file: Path, *, dry_run: bool = False) -> None:
"""Main.
Args:
config_file (Path): The path to the generated dataset declaration.
dry_run (bool): Log the changes that would be made without making them.
"""
configure_logger(level="DEBUG")
logger.info(f"Starting zfs_manager {dry_run=}")
try:
reconcile(config_file, dry_run=dry_run)
except ReconciliationError as error:
summary = error
except Exception:
logger.exception("zfs_manager failed")
signal_alert("zfs_manager failed")
sys.exit(1)
else:
logger.info("zfs_manager completed")
return
# Each failure was logged and alerted as it happened. Repeating them
# together puts the whole picture at the end of the journal, which is what
# systemctl status shows. No traceback: this is an expected outcome, not a
# crash, and a stack trace would only bury the list.
logger.error(str(summary))
for failure in summary.failures:
logger.error(f" {failure}")
sys.exit(1)
def reconcile(config_file: Path, *, dry_run: bool) -> None:
"""Bring every declared dataset in line, collecting problems as it goes.
One bad dataset must not hide the state of the others, so everything is
checked before anything is raised.
Args:
config_file (Path): The path to the generated dataset declaration.
dry_run (bool): Log the changes without making them.
Raises:
ReconciliationError: If anything could not be reconciled.
"""
declared = json.loads(config_file.read_text())["datasets"]
existing = set(list_dataset_names())
unusable: set[str] = set()
failures: list[str] = []
# Parents before children so a newly created parent exists by the time its
# children are reconciled.
for name in sorted(declared, key=lambda name: (name.count("/"), name)):
entry = declared[name]
# Declared purely to record retention, its properties belong to
# whoever set them.
if not entry.get("manageProperties", True):
logger.debug(f"{name} is declared but its properties are not managed")
continue
if has_unusable_parent(name, unusable):
failures.append(fail(f"cannot reconcile {name}, its parent is missing"))
continue
if name in existing:
failures.extend(reconcile_dataset(name, entry["properties"], dry_run=dry_run))
continue
created, failure = handle_missing_dataset(name, entry, dry_run=dry_run)
if failure is not None:
failures.append(failure)
if created:
existing.add(name)
elif not dry_run:
unusable.add(name)
report_undeclared_datasets(existing, declared)
if failures:
raise ReconciliationError(failures)
def fail(message: str) -> str:
"""Log and alert a problem, and hand it back for the failure tally.
Args:
message (str): What went wrong.
Returns:
str: The same message, so the caller can collect it.
"""
logger.error(message)
signal_alert(message)
return message
def has_unusable_parent(name: str, unusable: set[str]) -> bool:
"""Check whether an ancestor of a dataset is missing.
Args:
name (str): The name of the dataset.
unusable (set[str]): The datasets that do not exist and were not created.
Returns:
bool: True if any ancestor is unusable.
"""
parts = name.split("/")
return any("/".join(parts[:depth]) in unusable for depth in range(1, len(parts)))
def handle_missing_dataset(name: str, entry: dict, *, dry_run: bool) -> tuple[bool, str | None]:
"""Deal with a declared dataset that is not on the system.
Pool roots are never created, and neither is anything the declaration marks
as provisioned outside of nix, such as an encryption root whose key
settings cannot be reproduced from the declaration.
Args:
name (str): The name of the dataset.
entry (dict): The declaration for this dataset.
dry_run (bool): Log the change without making it.
Returns:
tuple[bool, str | None]: Whether the dataset now exists, and a failure
message if there was one.
"""
properties = entry["properties"]
if "/" not in name:
return False, fail(f"pool {name} is declared but does not exist, zfs_manager does not create pools")
if not entry.get("createIfMissing", True):
return False, fail(
f"{name} is declared but does not exist, and is marked as created outside of nix. "
"It has to be made by hand, see systems/jeeves/scripts/zfs.sh.",
)
if dry_run:
logger.info(f"would create {name} with {properties}")
return False, None
logger.info(f"creating {name} with {properties}")
if error := create_dataset(name, properties):
return False, fail(error)
return True, None
def reconcile_dataset(name: str, properties: dict[str, str], *, dry_run: bool) -> list[str]:
"""Bring an existing dataset in line with its declared properties.
Args:
name (str): The name of the dataset.
properties (dict[str, str]): The declared properties.
dry_run (bool): Log the changes without making them.
Returns:
list[str]: Anything that could not be put right.
"""
failures: list[str] = []
current = get_properties(name)
for key, wanted in sorted(properties.items()):
current_value, _ = current.get(key, ("-", "-"))
if values_match(key, wanted, current_value):
continue
if key in CREATE_ONLY_PROPERTIES:
# Nothing can put this right while the dataset exists, so it is a
# hard failure rather than a warning that repeats unnoticed.
failures.append(
fail(
f"{name} {key} is {current_value} but {wanted} is declared, "
f"{key} can only be set when the dataset is created",
),
)
continue
if dry_run:
logger.info(f"would set {key}={wanted} on {name}, currently {current_value}")
continue
logger.info(f"setting {key}={wanted} on {name}, was {current_value}")
if error := set_property(name, key, wanted):
failures.append(fail(error))
report_undeclared_properties(name, properties, current)
return failures
def report_undeclared_properties(name: str, properties: dict[str, str], current: dict[str, tuple[str, str]]) -> None:
"""Warn about properties set on the dataset but absent from the declaration.
Inherited and default values are silent, they are not drift. A locally set
value that nix does not know about was changed outside of this tool and
will be lost the next time the dataset is recreated, so it is worth saying.
Args:
name (str): The name of the dataset.
properties (dict[str, str]): The declared properties.
current (dict[str, tuple[str, str]]): The live properties keyed to (value, source).
"""
for key, (value, source) in sorted(current.items()):
# User properties such as nixos:shutdown-time are written by other
# tools and are not something a dataset declaration should own.
if key in properties or ":" in key or source not in LOCAL_SOURCES:
continue
logger.warning(f"{name} has {key}={value} set outside of nix")
signal_alert(f"{name} has {key}={value} set outside of nix")
def report_undeclared_datasets(existing: set[str], declared: dict[str, dict]) -> None:
"""Warn about datasets that exist but are not declared.
These are left completely alone. They still get snapshots through the
default retention table.
Args:
existing (set[str]): The names of every live dataset.
declared (dict[str, dict]): The declaration.
"""
for name in sorted(existing - set(declared)):
logger.warning(f"{name} exists but is not declared in nix")
def values_match(key: str, wanted: str, current: str) -> bool:
"""Compare a declared property value against the live one.
Args:
key (str): The property name.
wanted (str): The declared value.
current (str): The live value.
Returns:
bool: True if the two values mean the same thing.
"""
if key in SIZE_PROPERTIES:
wanted_size = parse_size(wanted)
current_size = parse_size(current)
if wanted_size is not None and current_size is not None:
return wanted_size == current_size
return wanted == current
def parse_size(value: str) -> int | None:
"""Convert a zfs size such as 16k or 1M into bytes.
Args:
value (str): The size to convert.
Returns:
int | None: The size in bytes, or None if it is not a size.
"""
value = value.strip()
if value.isdigit():
return int(value)
number, suffix = value[:-1], value[-1:].lower()
if suffix in SIZE_SUFFIXES and number.isdigit():
return int(number) * SIZE_SUFFIXES[suffix]
return None
def cli() -> None:
"""CLI."""
typer.run(main)
if __name__ == "__main__":
cli()