tools/sp-rand/src/sp_rand/cleanup.py

"""Clean up: remove volumes, snapshots, and the file itself."""

from __future__ import annotations

import argparse
import errno
import pathlib

from typing import NamedTuple, Tuple  # noqa: H301

import confget

from storpool import spapi  # type: ignore
from storpool import sptypes

from . import defs


class Config(NamedTuple):
    """Runtime configuration for the sp_rand_cleanup tool."""

    confdir: pathlib.Path
    noop: bool
    skip_api: bool

    @staticmethod
    def default() -> Config:
        """Build a configuration object with default values."""
        return Config(confdir=defs.DEFAULT_CONFDIR, noop=False, skip_api=False)


def parse_file(cfg: Config) -> Tuple[pathlib.Path, str]:
    """Parse the StorPool configuration file."""
    conffile = cfg.confdir / defs.FILENAME
    print(f"Parsing {conffile}")
    data = confget.read_ini_file(confget.Config([], filename=str(conffile)))

    assert isinstance(data, dict)
    assert sorted(data) == [""]
    assert sorted(data[""]) == [defs.PREFIX_VAR]
    prefix = data[""][defs.PREFIX_VAR]
    print(f"- got prefix {prefix!r}")

    return conffile, prefix


def remove_volumes(cfg: Config, api: spapi.Api, prefix: str) -> None:
    """Detach and remove any pertinent StorPool volumes and snapshots."""
    print("Querying the StorPool API for attached volumes and snapshots")
    to_detach = []
    for desc in api.attachmentsList():  # pylint: disable=not-callable
        if not desc.volume.startswith(prefix):
            continue
        if desc.snapshot:
            print(f"Will detach snapshot {desc.volume} from client {desc.client}")
            to_detach.append(
                sptypes.SnapshotReassignDesc(
                    detach=[desc.client], force=True, snapshot=desc.volume
                )
            )
        else:
            print(f"Will detach volume {desc.volume} from client {desc.client}")
            to_detach.append(
                sptypes.VolumeReassignDesc(
                    detach=[desc.client], force=True, volume=desc.volume
                )
            )

    if to_detach:
        print(f"Detaching {len(to_detach)} volumes/snapshots")
        if not cfg.noop:
            api.volumesReassignWait(  # pylint: disable=not-callable
                sptypes.VolumesReassignWaitDesc(reassign=to_detach)
            )

    print("Querying the StorPool API for existing volumes")
    for name in [vol.name for vol in api.volumesList()]:
        if not name.startswith(prefix):
            continue
        print(f"Removing volume {name}")
        if not cfg.noop:
            api.volumeDelete(name)

    print("Querying the StorPool API for existing snapshots")
    for name in [snap.name for snap in api.snapshotsList()]:
        if not name.startswith(prefix):
            continue
        print(f"Removing snapshot {name}")
        if not cfg.noop:
            api.snapshotDelete(name)


def remove_file(cfg: Config, conffile: pathlib.Path) -> None:
    """Remove the StorPool configuration file."""
    print(f"Removing {conffile}")
    if not cfg.noop:
        try:
            conffile.unlink()
        except OSError as err:
            if err.errno != errno.ENOENT:
                raise


def parse_args() -> Config:
    """Parse the command-line arguments."""
    parser = argparse.ArgumentParser(prog="sp_rand_cleanup")
    parser.add_argument(
        "-d",
        "--confdir",
        type=pathlib.Path,
        default=defs.DEFAULT_CONFDIR,
        help="The directory to look for the configuration file in",
    )
    parser.add_argument(
        "-N",
        "--noop",
        action="store_true",
        help="No-operation mode; display what would have been done",
    )
    parser.add_argument(
        "--skip-api",
        type=str,
        help=(
            "If explicitly given the 'yes' value, do not "
            "delete any volumes or snapshots"
        ),
    )

    args = parser.parse_args()

    return Config(confdir=args.confdir, noop=args.noop, skip_api=args.skip_api == "yes")


def main() -> None:
    """Main program: parse command-line options, handle them."""
    cfg = parse_args()
    conffile, prefix = parse_file(cfg)

    if not cfg.skip_api:
        api = spapi.Api.fromConfig()
        remove_volumes(cfg, api, prefix)

    remove_file(cfg, conffile)


if __name__ == "__main__":
    main()