ansible/bin/resize_filesystem

#!/usr/bin/env python3
"""Resize an LVM logical volume and its filesystem.

Author: Bart Sjerps
Date: 2026-07-17
Generated by opencode
"""
import os
import sys
import subprocess
import argparse
from typing import Optional


def die(msg: str) -> None:
    """Print an error message to stderr and exit with code 2."""
    print(f"Error: {msg}", file=sys.stderr)
    sys.exit(2)


def run(cmd: list[str], dryrun: bool = False) -> str:
    """Execute a shell command, returning stdout. Exits on non-zero status."""
    if dryrun:
        print(f"[DRY-RUN] {' '.join(cmd)}")
        return ""
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    except subprocess.CalledProcessError as e:
        err = (e.stderr or e.stdout or "").strip()
        die(f"{' '.join(cmd)} failed: {err or f'exit code {e.returncode}'}")
    return result.stdout.strip()


def resolve_lv(mount: Optional[str] = None, lvname: Optional[str] = None) -> str:
    """Resolve a mountpoint or LV name to an absolute LVM device path."""
    if mount:
        dev = run(["findmnt", "-no", "SOURCE", "--target", mount])
        if not dev:
            die(f"No device mounted at {mount}")
    else:
        dev = lvname

    if not dev.startswith("/dev/"):
        dev = f"/dev/{dev}"

    if not run(["lvs", dev]):
        die(f"{dev} is not an LVM logical volume")

    return dev


def detect_fs(dev: str) -> str:
    """Detect the filesystem type on a block device via blkid."""
    fstype = run(["blkid", "-o", "value", "-s", "TYPE", dev])
    if not fstype:
        die(f"Cannot detect filesystem on {dev}")
    return fstype


def resolve_mount(dev: str, mount: Optional[str] = None) -> str:
    """Return a mountpoint for a device, using the provided one or looking it up."""
    if mount:
        return mount

    mountpoint = run(["findmnt", "-no", "TARGET", "--source", dev])
    if not mountpoint:
        die(f"No mountpoint found for {dev}")
    return mountpoint


def lvextend_cmd(size: str, dev: str) -> list[str]:
    """Build an lvextend command, using -l for extent/percent sizes."""
    if "%" in size:
        return ["lvextend", "-l", size, dev]
    return ["lvextend", "-L", size, dev]


def btrfs_resize_size(size: str) -> str:
    """Map an LVM size spec to a btrfs filesystem resize argument."""
    if "%" in size:
        return "max"
    return size


def validate_resize(dev: str, fstype: str, mount: Optional[str] = None) -> Optional[str]:
    """Validate resize is possible before mutating the LV. Returns mountpoint when needed."""
    if fstype in ("ext2", "ext3", "ext4"):
        return None
    if fstype in ("xfs", "btrfs"):
        return resolve_mount(dev, mount)
    die(f"Unsupported filesystem: {fstype}")


def do_resize(dev: str, fstype: str, size: str, mount: Optional[str] = None, dryrun: bool = False) -> None:
    """Extend the LV with lvextend, then resize the filesystem accordingly."""
    fs_mount = validate_resize(dev, fstype, mount)

    run(lvextend_cmd(size, dev), dryrun=dryrun)

    if fstype in ("ext2", "ext3", "ext4"):
        run(["resize2fs", dev], dryrun=dryrun)
    elif fstype == "xfs":
        run(["xfs_growfs", fs_mount], dryrun=dryrun)
    elif fstype == "btrfs":
        run(["btrfs", "filesystem", "resize", btrfs_resize_size(size), fs_mount], dryrun=dryrun)

    if not dryrun:
        print("Done.")


def main() -> None:
    """Parse args, validate root, then resize LV and filesystem."""
    parser = argparse.ArgumentParser(description="Resize an LVM logical volume and its filesystem")
    parser.add_argument("-s", dest="size", required=True, help="New size (e.g. +10G, -5G, 20G, 100%FREE)")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-m", dest="mount", metavar="MOUNTPOINT", help="Select LV by mountpoint")
    group.add_argument("-l", dest="lvname", metavar="LVNAME", help="Select LV by name or path (e.g. vg0/data)")
    parser.add_argument("-n", dest="dryrun", action="store_true", help="Dry-run – print what would be done without executing")

    args = parser.parse_args()

    if os.geteuid() != 0:
        die("This script must be run as root")

    lvdev = resolve_lv(mount=args.mount, lvname=args.lvname)
    fstype = detect_fs(lvdev)
    do_resize(lvdev, fstype, args.size, mount=args.mount, dryrun=args.dryrun)


if __name__ == "__main__":
    main()