btrfs-scrub-individual.py

btrfs-scrub-individual.py

#! /usr/bin/env python3

import argparse
import re
import subprocess
import sys
from typing import List, NoReturn, Set

PATH_REGEX = re.compile("path\s+(?P<path>[^\s]+)")


def get_scrub_flags(args: argparse.Namespace) -> List[str]:
    FLAGS = ["-c", "-n", "-r", "-R", "-d"]

    results = []
    args_dict = vars(args)

    for flag in FLAGS:
        if flag[1:] not in args_dict:
            continue

        value = args_dict[flag[1:]]
        if isinstance(value, bool):
            # Only add conditional flags if they are True
            if value:
                results.append(flag)
        else:
            results.append(f"{flag} {value}")

    # Always add the "no background" flag
    results.append("-B")

    return results


def get_devices(id: str) -> Set[str]:
    cmd_result = subprocess.run(
        ["btrfs", "filesystem", "show", id], capture_output=True
    )

    if cmd_result.returncode != 0:
        sys.stdout.buffer.write(cmd_result.stdout)
        sys.stderr.buffer.write(cmd_result.stderr)
        exit(cmd_result.returncode)

    output = cmd_result.stdout.decode("utf-8")

    return set(match.group("path") for match in PATH_REGEX.finditer(output))


def build_command(device: str, flags: Set[str]):
    return ["btrfs", "scrub", "start"] + flags + [device]


def scrub_dry_run(devices: List[str], flags: Set[str]) -> NoReturn:
    for device in devices:
        cmd = build_command(device, flags)
        print(" ".join(cmd))

    exit()


def scrub_run(devices: List[str], flags: Set[str]) -> NoReturn:
    exit_code = 0
    for device in devices:
        cmd = build_command(device, flags)

        cmd_result = subprocess.run(cmd)

        if cmd_result.returncode != 0 and exit_code == 0:
            exit_code = cmd_result.returncode

    exit(exit_code)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        "btrfs-scrub-individual",
        description="Scrub each individual device in btrfs separately",
    )
    parser.add_argument(
        "id",
        type=str,
        help="the path, UUID, device, or label of the btrfs FS to scrub",
    )
    parser.add_argument(
        "-c",
        nargs=1,
        type=int,
        help="set IO priority class (see ionice(1) manpage)",
    )
    parser.add_argument(
        "-n",
        action="store_true",
        help="set IO priority classdata (see ionice(1) manpage)",
    )
    parser.add_argument(
        "-r",
        action="store_true",
        help="run in read-only mode, do not attempt to correct anything, can be run on a read-only filesystem",
    )
    parser.add_argument(
        "-R",
        action="store_true",
        help="raw print mode, print full data instead of summary",
    )
    parser.add_argument(
        "-d",
        action="store_true",
        help="print separate statistics for each device of the filesystem at the end",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="simulate a run and output the scrub commands that would've run",
    )

    return parser.parse_args()


def main():
    args = parse_args()

    id = str(args.id)

    devices = get_devices(id)
    flags = get_scrub_flags(args)

    if args.dry_run:
        scrub_fn = scrub_dry_run
    else:
        scrub_fn = scrub_run

    scrub_fn(devices, flags)


if __name__ == "__main__":
    main()