Richie
48a7e3a54c
treefmt / nix fmt (pull_request) Successful in 6s
pytest / pytest (pull_request) Successful in 30s
test ebook search / test-ebook-search (pull_request) Successful in 36s
build_systems / build-brain (pull_request) Successful in 46s
build_systems / build-bob (pull_request) Successful in 47s
build_systems / build-rhapsody-in-green (pull_request) Successful in 58s
build_systems / build-jeeves (pull_request) Successful in 2m23s
pytest / pytest (push) Successful in 33s
test ebook search / test-ebook-search (push) Successful in 42s
build_systems / build-jeeves (push) Successful in 2m26s
treefmt / nix fmt (push) Successful in 5s
build_systems / build-brain (push) Successful in 9s
build_systems / build-bob (push) Successful in 40s
build_systems / build-rhapsody-in-green (push) Successful in 53s
299 lines
11 KiB
Python
299 lines
11 KiB
Python
"""dataset."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from python.common import bash_wrapper
|
|
from python.zfs.command import run_zfs
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _zfs_list(zfs_list: str) -> dict[str, Any]:
|
|
"""Check the version of zfs."""
|
|
raw_zfs_list_data, _ = bash_wrapper(zfs_list)
|
|
|
|
zfs_list_data = json.loads(raw_zfs_list_data)
|
|
|
|
vers_major = zfs_list_data["output_version"]["vers_major"]
|
|
vers_minor = zfs_list_data["output_version"]["vers_minor"]
|
|
command = zfs_list_data["output_version"]["command"]
|
|
|
|
if vers_major != 0 or vers_minor != 1 or command != "zfs list":
|
|
error = f"Datasets are not in the correct format {vers_major=} {vers_minor=} {command=}"
|
|
raise RuntimeError(error)
|
|
|
|
return zfs_list_data
|
|
|
|
|
|
class Snapshot:
|
|
"""Snapshot."""
|
|
|
|
def __init__(self, snapshot_data: dict[str, Any]) -> None:
|
|
"""__init__."""
|
|
properties = snapshot_data["properties"]
|
|
self.createtxg = int(snapshot_data["createtxg"])
|
|
self.creation = datetime.fromtimestamp(int(properties["creation"]["value"]), tz=UTC)
|
|
self.defer_destroy = properties["defer_destroy"]["value"]
|
|
self.guid = int(properties["guid"]["value"])
|
|
self.name = snapshot_data["name"].split("@")[1]
|
|
self.objsetid = int(properties["objsetid"]["value"])
|
|
self.referenced = int(properties["referenced"]["value"])
|
|
self.used = int(properties["used"]["value"])
|
|
self.userrefs = int(properties["userrefs"]["value"])
|
|
self.version = int(properties["version"]["value"])
|
|
self.written = int(properties["written"]["value"])
|
|
|
|
def __repr__(self) -> str:
|
|
"""__repr__."""
|
|
return f"name={self.name} used={self.used} refer={self.referenced}"
|
|
|
|
|
|
class Dataset:
|
|
"""Dataset."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
"""__init__."""
|
|
dataset_data = _zfs_list(f"zfs list {name} -pHj -o all")
|
|
|
|
properties = dataset_data["datasets"][name]["properties"]
|
|
|
|
self.aclinherit = properties["aclinherit"]["value"]
|
|
self.aclmode = properties["aclmode"]["value"]
|
|
self.acltype = properties["acltype"]["value"]
|
|
self.available = int(properties["available"]["value"])
|
|
self.canmount = properties["canmount"]["value"]
|
|
self.checksum = properties["checksum"]["value"]
|
|
self.clones = properties["clones"]["value"]
|
|
self.compression = properties["compression"]["value"]
|
|
self.copies = int(properties["copies"]["value"])
|
|
self.createtxg = int(properties["createtxg"]["value"])
|
|
self.creation = datetime.fromtimestamp(int(properties["creation"]["value"]), tz=UTC)
|
|
self.dedup = properties["dedup"]["value"]
|
|
self.devices = properties["devices"]["value"]
|
|
self.encryption = properties["encryption"]["value"]
|
|
self.exec = properties["exec"]["value"]
|
|
self.filesystem_limit = properties["filesystem_limit"]["value"]
|
|
self.guid = int(properties["guid"]["value"])
|
|
self.keystatus = properties["keystatus"]["value"]
|
|
self.logbias = properties["logbias"]["value"]
|
|
self.mlslabel = properties["mlslabel"]["value"]
|
|
self.mounted = properties["mounted"]["value"]
|
|
self.mountpoint = properties["mountpoint"]["value"]
|
|
self.name = name
|
|
self.quota = int(properties["quota"]["value"])
|
|
self.readonly = properties["readonly"]["value"]
|
|
self.recordsize = int(properties["recordsize"]["value"])
|
|
self.redundant_metadata = properties["redundant_metadata"]["value"]
|
|
self.referenced = int(properties["referenced"]["value"])
|
|
self.refquota = int(properties["refquota"]["value"])
|
|
self.refreservation = int(properties["refreservation"]["value"])
|
|
self.reservation = int(properties["reservation"]["value"])
|
|
self.setuid = properties["setuid"]["value"]
|
|
self.sharenfs = properties["sharenfs"]["value"]
|
|
self.snapdir = properties["snapdir"]["value"]
|
|
self.snapshot_limit = properties["snapshot_limit"]["value"]
|
|
self.sync = properties["sync"]["value"]
|
|
self.used = int(properties["used"]["value"])
|
|
self.usedbychildren = int(properties["usedbychildren"]["value"])
|
|
self.usedbydataset = int(properties["usedbydataset"]["value"])
|
|
self.usedbysnapshots = int(properties["usedbysnapshots"]["value"])
|
|
self.version = int(properties["version"]["value"])
|
|
self.volmode = properties["volmode"]["value"]
|
|
self.volsize = properties["volsize"]["value"]
|
|
self.vscan = properties["vscan"]["value"]
|
|
self.written = int(properties["written"]["value"])
|
|
self.xattr = properties["xattr"]["value"]
|
|
|
|
def get_snapshots(self) -> list[Snapshot]:
|
|
"""Get all snapshots from zfs and process then is test dicts of sets."""
|
|
snapshots_data = _zfs_list(f"zfs list -t snapshot -pHj {self.name} -o all")
|
|
|
|
return [Snapshot(properties) for properties in snapshots_data["datasets"].values()]
|
|
|
|
def create_snapshot(self, snapshot_name: str) -> str:
|
|
"""Creates a zfs snapshot.
|
|
|
|
Args:
|
|
snapshot_name (str): a snapshot name
|
|
"""
|
|
logger.debug(f"Creating {self.name}@{snapshot_name}")
|
|
_, return_code = bash_wrapper(f"zfs snapshot {self.name}@{snapshot_name}")
|
|
if return_code == 0:
|
|
return "snapshot created"
|
|
|
|
snapshots = self.get_snapshots()
|
|
snapshot_names = {snapshot.name for snapshot in snapshots}
|
|
if snapshot_name in snapshot_names:
|
|
return f"Snapshot {snapshot_name} already exists for {self.name}"
|
|
|
|
return f"Failed to create snapshot {snapshot_name} for {self.name}"
|
|
|
|
def delete_snapshot(self, snapshot_name: str) -> str | None:
|
|
"""Deletes a zfs snapshot.
|
|
|
|
Args:
|
|
snapshot_name (str): a snapshot name
|
|
"""
|
|
logger.debug(f"deleting {self.name}@{snapshot_name}")
|
|
msg, return_code = bash_wrapper(f"zfs destroy {self.name}@{snapshot_name}")
|
|
if return_code != 0:
|
|
if msg.startswith(f"cannot destroy '{self.name}@{snapshot_name}': snapshot has dependent clones"):
|
|
return "snapshot has dependent clones"
|
|
error = f"Failed to delete snapshot {snapshot_name=} for {self.name}"
|
|
raise RuntimeError(error)
|
|
return None
|
|
|
|
def __repr__(self) -> str:
|
|
"""__repr__."""
|
|
return (
|
|
f"{self.aclinherit=}\n"
|
|
f"{self.aclmode=}\n"
|
|
f"{self.acltype=}\n"
|
|
f"{self.available=}\n"
|
|
f"{self.canmount=}\n"
|
|
f"{self.checksum=}\n"
|
|
f"{self.clones=}\n"
|
|
f"{self.compression=}\n"
|
|
f"{self.copies=}\n"
|
|
f"{self.createtxg=}\n"
|
|
f"{self.creation=}\n"
|
|
f"{self.dedup=}\n"
|
|
f"{self.devices=}\n"
|
|
f"{self.encryption=}\n"
|
|
f"{self.exec=}\n"
|
|
f"{self.filesystem_limit=}\n"
|
|
f"{self.guid=}\n"
|
|
f"{self.keystatus=}\n"
|
|
f"{self.logbias=}\n"
|
|
f"{self.mlslabel=}\n"
|
|
f"{self.mounted=}\n"
|
|
f"{self.mountpoint=}\n"
|
|
f"{self.name=}\n"
|
|
f"{self.quota=}\n"
|
|
f"{self.readonly=}\n"
|
|
f"{self.recordsize=}\n"
|
|
f"{self.redundant_metadata=}\n"
|
|
f"{self.referenced=}\n"
|
|
f"{self.refquota=}\n"
|
|
f"{self.refreservation=}\n"
|
|
f"{self.reservation=}\n"
|
|
f"{self.setuid=}\n"
|
|
f"{self.sharenfs=}\n"
|
|
f"{self.snapdir=}\n"
|
|
f"{self.snapshot_limit=}\n"
|
|
f"{self.sync=}\n"
|
|
f"{self.used=}\n"
|
|
f"{self.usedbychildren=}\n"
|
|
f"{self.usedbydataset=}\n"
|
|
f"{self.usedbysnapshots=}\n"
|
|
f"{self.version=}\n"
|
|
f"{self.volmode=}\n"
|
|
f"{self.volsize=}\n"
|
|
f"{self.vscan=}\n"
|
|
f"{self.written=}\n"
|
|
f"{self.xattr=}\n"
|
|
)
|
|
|
|
|
|
def get_datasets() -> list[Dataset]:
|
|
"""Get zfs list.
|
|
|
|
Returns:
|
|
list[Dataset]: A list of zfs datasets.
|
|
"""
|
|
logger.info("Getting zfs list")
|
|
|
|
return [Dataset(dataset_name) for dataset_name in list_dataset_names() if "/" in dataset_name]
|
|
|
|
|
|
def get_properties(name: str) -> dict[str, tuple[str, str]]:
|
|
"""Get every property of a dataset along with where its value came from.
|
|
|
|
The source is what distinguishes a property that was deliberately set on
|
|
this dataset from one that is merely inherited or left at its default.
|
|
|
|
Args:
|
|
name (str): The name of the dataset.
|
|
|
|
Returns:
|
|
dict[str, tuple[str, str]]: A mapping of property name to (value, source).
|
|
"""
|
|
result = run_zfs("get", "-Hp", "-o", "property,value,source", "all", name)
|
|
if not result.ok:
|
|
error = f"Failed to get properties for {name}: {result.message}"
|
|
raise RuntimeError(error)
|
|
|
|
properties = {}
|
|
for line in result.stdout.strip().splitlines():
|
|
if not line:
|
|
continue
|
|
prop, value, source = line.split("\t")
|
|
properties[prop] = (value, source)
|
|
|
|
return properties
|
|
|
|
|
|
def create_dataset(name: str, properties: dict[str, str]) -> str | None:
|
|
"""Create a dataset with the given properties.
|
|
|
|
Args:
|
|
name (str): The name of the dataset.
|
|
properties (dict[str, str]): The properties to create the dataset with.
|
|
|
|
Returns:
|
|
str | None: An error message on failure, None on success.
|
|
"""
|
|
options = [argument for key, value in sorted(properties.items()) for argument in ("-o", f"{key}={value}")]
|
|
logger.debug(f"creating {name} with {properties}")
|
|
|
|
result = run_zfs("create", *options, name)
|
|
if not result.ok:
|
|
return f"Failed to create {name}: {result.message}"
|
|
return None
|
|
|
|
|
|
def set_property(name: str, key: str, value: str) -> str | None:
|
|
"""Set a single property on a dataset.
|
|
|
|
Args:
|
|
name (str): The name of the dataset.
|
|
key (str): The property to set.
|
|
value (str): The value to set the property to.
|
|
|
|
Returns:
|
|
str | None: An error message on failure, None on success.
|
|
"""
|
|
logger.debug(f"setting {key}={value} on {name}")
|
|
|
|
result = run_zfs("set", f"{key}={value}", name)
|
|
if not result.ok:
|
|
return f"Failed to set {key}={value} on {name}: {result.message}"
|
|
return None
|
|
|
|
|
|
def list_dataset_names() -> list[str]:
|
|
"""List every zfs filesystem name, including pool root filesystems.
|
|
|
|
Unlike get_datasets this does not build Dataset objects and does not filter
|
|
out pool roots, which makes it usable for existence checks.
|
|
|
|
Returns:
|
|
list[str]: The names of every zfs filesystem.
|
|
|
|
Raises:
|
|
RuntimeError: If zfs list fails. Never returns a partial or error
|
|
derived list, since treating stderr as dataset names would make
|
|
the reconciler think every dataset is missing.
|
|
"""
|
|
result = run_zfs("list", "-Hp", "-t", "filesystem", "-o", "name")
|
|
if not result.ok:
|
|
error = f"Failed to list ZFS datasets: {result.message}"
|
|
raise RuntimeError(error)
|
|
|
|
return [name for name in result.stdout.strip().splitlines() if name]
|