From e05bb3e6e225435e9709582c13011f24cde98d14 Mon Sep 17 00:00:00 2001 From: Zack Cerza Date: Wed, 21 Oct 2026 14:30:33 -0600 Subject: [PATCH 1/3] Add a shared block pool for loop devices Loop devices can now be backed by slices of a dedicated disk or partition instead of sparse files. - BlockPool enrolls an empty device and hands out fixed-size regions, tracked in a locked state file. It refuses devices that aren't empty or aren't its own. - Released regions are zeroed wherever BlueStore keeps a label copy; otherwise ceph-volume silently skips a reused region. - LoopDeviceMixin picks free loop numbers, safely across parallel creates, and attaches them to pool regions or sparse files. Device access runs on the container host, so this also works through a podman machine. Signed-off-by: Zack Cerza --- ceph_devstack/__init__.py | 1 + ceph_devstack/block_pool.py | 474 ++++++++++++++++++ ceph_devstack/exec.py | 7 + ceph_devstack/host.py | 77 ++- ceph_devstack/resources/ceph/block_devices.py | 254 ++++++++++ ceph_devstack/resources/ceph/host_loops.py | 123 +++++ tests/conftest.py | 150 +++++- tests/resources/ceph/conftest.py | 50 ++ tests/resources/ceph/test_block_devices.py | 226 +++++++++ tests/resources/ceph/test_host_loops.py | 117 +++++ tests/test_block_pool.py | 330 ++++++++++++ tests/test_host.py | 81 +++ 12 files changed, 1886 insertions(+), 4 deletions(-) create mode 100644 ceph_devstack/block_pool.py create mode 100644 ceph_devstack/resources/ceph/block_devices.py create mode 100644 ceph_devstack/resources/ceph/host_loops.py create mode 100644 tests/resources/ceph/conftest.py create mode 100644 tests/resources/ceph/test_block_devices.py create mode 100644 tests/resources/ceph/test_host_loops.py create mode 100644 tests/test_block_pool.py diff --git a/ceph_devstack/__init__.py b/ceph_devstack/__init__.py index d71474cd..9eae8d52 100644 --- a/ceph_devstack/__init__.py +++ b/ceph_devstack/__init__.py @@ -15,6 +15,7 @@ PROJECT_ROOT = Path(__file__).parent DEFAULT_CONFIG_PATH = Path("~/.config/ceph-devstack/config.toml") +DEFAULT_DATA_DIR = "~/.local/share/ceph-devstack" def parse_args(args: List[str]) -> argparse.Namespace: diff --git a/ceph_devstack/block_pool.py b/ceph_devstack/block_pool.py new file mode 100644 index 00000000..f4890dac --- /dev/null +++ b/ceph_devstack/block_pool.py @@ -0,0 +1,474 @@ +import contextlib +import fcntl +import functools +import json +import os +import re +import uuid +from pathlib import Path + +from ceph_devstack import DEFAULT_DATA_DIR, config +from ceph_devstack.host import host + + +_SIZE_RE = re.compile(r"^(\d+(?:\.\d+)?)([KMGTP]?)$", re.IGNORECASE) +_UNITS = { + "": 1, + "K": 1024, + "M": 1024**2, + "G": 1024**3, + "T": 1024**4, + "P": 1024**5, +} +_SECTOR_SIZE = 512 +_SYSFS_BLOCK = Path("/sys/class/block") + +# Written to the last 4 KiB of an enrolled parent so we can recognize our own +# partitions even if block_pool.json is removed. +_POOL_TAIL_MAGIC = b"CEPH-DEVSTACK-BLOCK-POOL-TAIL-v1\x00" +_TAIL_MARKER_SIZE = 4096 +_PROBE_SIZE = 4096 +# BlueStore writes its label at the start of a device and copies of it at +# these offsets; ceph-volume refuses (silently, for raw prepare) a device +# that carries any of them. Region probes cover each one inside the region. +_BLUESTORE_LABEL_OFFSETS = (0, 1024**3, 10 * 1024**3, 100 * 1024**3, 1000 * 1024**3) + +# Whole disks and partitions are allowed; device-mapper/loop types are not. +# NVMe and MMC partitions add "p" to the disk name; SCSI/virtio add "". +_NVME_DISK = r"nvme\d+n\d+" +_MMC_DISK = r"mmcblk\d+" +_SCSI_DISK = r"(?:sd|[xv]d)[a-z]+" +_WHOLE_DISK_NAME = re.compile(rf"^(?:{_NVME_DISK}|{_MMC_DISK}|{_SCSI_DISK})$") +_PARTITION_NAME = re.compile(rf"^(?:(?:{_NVME_DISK}|{_MMC_DISK})p|{_SCSI_DISK})\d+$") +_UNSAFE_NAME = re.compile(r"^(?:dm-\d+|md\d+|loop\d+|ram\d+|fd\d+)$") + + +class BlockPoolError(ValueError): + """Raised when a block pool operation would be unsafe.""" + + +def parse_size(value: str | int) -> int: + text = str(value).strip().upper() + match = _SIZE_RE.match(text) + if not match: + raise BlockPoolError(f"invalid size: {value!r}") + number, unit = match.groups() + size = int(float(number) * _UNITS[unit.upper()]) + if size <= 0: + raise BlockPoolError(f"size must be positive: {value!r}") + return size + + +def format_size(size_bytes: int) -> str: + for unit in ("P", "T", "G", "M", "K"): + multiplier = _UNITS[unit] + if size_bytes >= multiplier and size_bytes % multiplier == 0: + return f"{size_bytes // multiplier}{unit}" + return f"{size_bytes}B" + + +def validate_parent_name(parent: str) -> str: + """Return canonical parent path or raise if the device is not a safe block device.""" + parent = host.realpath(parent) + if not parent.startswith("/dev/"): + raise BlockPoolError(f"block pool parent must be a /dev path, not {parent!r}") + name = os.path.basename(parent) + if _UNSAFE_NAME.match(name): + raise BlockPoolError( + f"block pool parent {parent!r} is not allowed " + "(device-mapper, md, loop, and ram devices are rejected)" + ) + if not (_PARTITION_NAME.match(name) or _WHOLE_DISK_NAME.match(name)): + raise BlockPoolError( + f"block pool parent {parent!r} must be a block device or partition " + "(e.g. /dev/nvme0n1, /dev/nvme0n1p1, or /dev/sda1)" + ) + if not host.path_exists(parent): + raise BlockPoolError(f"block pool parent {parent!r} does not exist") + if not host.path_exists(_SYSFS_BLOCK / name): + raise BlockPoolError(f"block pool parent {parent!r} is not a block device") + return parent + + +class BlockPool: + """Allocate sized regions from a shared parent block device. + + Device access (sysfs, findmnt, blkid, marker and probe I/O) goes through + the Host abstraction; the state file and its lock stay local. + """ + + @staticmethod + def _has_partitions(name: str) -> bool: + """True when a whole-disk parent already has partitions. + + The kernel lists a disk's partitions as subdirectories of its sysfs + directory, so this needs no name-prefix guessing. + """ + if not _WHOLE_DISK_NAME.match(name): + return False + return any(_PARTITION_NAME.match(e) for e in host.list_dir(_SYSFS_BLOCK / name)) + + @functools.cached_property + def _device_size(self) -> int: + name = os.path.basename(self.parent) + sectors = host.read_text(_SYSFS_BLOCK / name / "size") + try: + return int(sectors or "") * _SECTOR_SIZE + except ValueError: + raise BlockPoolError(f"cannot read size of {self.parent!r}") from None + + def _read_parent(self, offset: int, length: int) -> bytes: + try: + return host.read_bytes(self.parent, offset, length) + except OSError as e: + raise BlockPoolError(str(e)) from e + + def _write_parent(self, offset: int, data: bytes) -> None: + try: + host.write_bytes(self.parent, offset, data) + except OSError as e: + raise BlockPoolError(str(e)) from e + + @staticmethod + def _is_region_empty(data: bytes) -> bool: + return not data or data == b"\x00" * len(data) + + def _device_has_blkid_signature(self) -> bool: + proc = host.output(["blkid", "-p", "-o", "export", self.parent]) + if proc.returncode != 0: + return False + for line in proc.stdout.decode().splitlines(): + key, _, value = line.partition("=") + if key in {"TYPE", "PTTYPE"} and value: + return True + return False + + def _tail_marker_offset(self) -> int: + size = self._device_size + if size < _TAIL_MARKER_SIZE: + raise BlockPoolError(f"block pool parent {self.parent!r} is too small") + return size - _TAIL_MARKER_SIZE + + def _read_tail_marker(self) -> dict | None: + offset = self._tail_marker_offset() + raw = self._read_parent(offset, _TAIL_MARKER_SIZE) + if not raw.startswith(_POOL_TAIL_MAGIC): + return None + payload = raw[len(_POOL_TAIL_MAGIC) :].split(b"\x00", 1)[0] + if not payload: + return None + try: + return json.loads(payload.decode()) + except json.JSONDecodeError: + return None + + def _write_tail_marker(self, payload: dict) -> None: + offset = self._tail_marker_offset() + encoded = json.dumps(payload, sort_keys=True).encode() + if len(_POOL_TAIL_MAGIC) + len(encoded) + 1 > _TAIL_MARKER_SIZE: + raise BlockPoolError("block pool tail marker payload is too large") + raw = _POOL_TAIL_MAGIC + encoded + b"\x00" + raw = raw.ljust(_TAIL_MARKER_SIZE, b"\x00") + self._write_parent(offset, raw) + + def _probe_parent_empty(self) -> None: + size = self._device_size + offsets = [0] + if size > _SECTOR_SIZE: + offsets.append(_SECTOR_SIZE) + if size > _PROBE_SIZE: + offsets.append(size // 2) + tail = self._tail_marker_offset() + if tail not in offsets and tail >= _PROBE_SIZE: + offsets.append(tail) + for offset in offsets: + data = self._read_parent(offset, _PROBE_SIZE) + if not self._is_region_empty(data): + raise BlockPoolError( + f"block pool parent {self.parent!r} is not empty at offset " + f"{offset}; refusing to enroll a partition that may contain data" + ) + + def __init__( + self, + state_path: Path, + parent: str, + *, + allow_enroll: bool = False, + ): + self.state_path = state_path + self.parent = validate_parent_name(parent) + self.allow_enroll = allow_enroll + self._state = self._load() + + @classmethod + def from_config(cls, cfg: dict) -> "BlockPool | None": + pool_cfg = cfg.get("block_pool") or {} + parent = pool_cfg.get("parent") + if not parent: + return None + state_dir = pool_cfg.get("state_dir") or cfg.get("data_dir", DEFAULT_DATA_DIR) + state_path = Path(state_dir).expanduser().absolute() / "block_pool.json" + allow_enroll = pool_cfg.get("allow_enroll", False) is True + return cls(state_path, parent, allow_enroll=allow_enroll) + + @property + def enrolled(self) -> bool: + return bool(self._state.get("enrolled")) + + @property + def pool_id(self) -> str: + return self._state["pool_id"] + + @property + def allocations(self) -> dict[str, dict]: + """Allocated regions keyed by ":".""" + return dict(self._state["allocations"]) + + def owner_offsets(self, owner: str) -> set[int]: + """Offsets of the regions currently allocated to owner (fresh from disk).""" + with self._state_lock(): + return { + region["offset"] + for region in self._state["allocations"].values() + if region["owner"] == owner + } + + @property + def free_regions(self) -> list[dict]: + return list(self._state["free_regions"]) + + @property + def next_offset(self) -> int: + return self._state["next_offset"] + + def _load(self) -> dict: + if self.state_path.exists(): + state = json.loads(self.state_path.read_text()) + if state.get("parent") != self.parent: + raise BlockPoolError( + f"block pool parent changed ({state.get('parent')!r} -> " + f"{self.parent!r}); remove {self.state_path} to reconfigure" + ) + return state + return self._fresh_state() + + def _fresh_state(self) -> dict: + return { + "parent": self.parent, + "pool_id": str(uuid.uuid4()), + "enrolled": False, + "next_offset": 0, + "free_regions": [], + "allocations": {}, + } + + @contextlib.contextmanager + def _state_lock(self): + self.state_path.parent.mkdir(parents=True, exist_ok=True) + lock_path = self.state_path.with_suffix(".lock") + with open(lock_path, "w") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + try: + if self.state_path.exists(): + self._state = self._load() + yield + finally: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + def _atomic_save(self) -> None: + tmp_path = self.state_path.with_suffix(".tmp") + tmp_path.write_text(json.dumps(self._state, indent=2, sort_keys=True) + "\n") + os.replace(tmp_path, self.state_path) + + def ensure_ready(self) -> None: + """Verify the parent partition is safe to use before any slice I/O.""" + with self._state_lock(): + if self._ensure_ready_unlocked(): + self._atomic_save() + + def _ensure_ready_unlocked(self) -> bool: + if self.enrolled: + self._verify_enrolled_parent() + return False + self._enroll_new_parent() + return True + + def _verify_enrolled_parent(self) -> None: + marker = self._read_tail_marker() + if marker is None: + raise BlockPoolError( + f"block pool parent {self.parent!r} is missing the devstack " + f"enrollment marker; refusing to use it. If {self.state_path} " + "was removed accidentally, start again to reclaim the marker. " + "If you wiped the device, set block_pool.allow_enroll = true " + "for a fresh enrollment." + ) + if marker.get("pool_id") != self._state.get("pool_id"): + raise BlockPoolError( + f"block pool parent {self.parent!r} belongs to another pool " + f"({marker.get('pool_id')!r}); remove {self.state_path} only " + "if you intentionally reset the pool" + ) + if marker.get("parent") != self.parent: + raise BlockPoolError( + f"block pool parent path mismatch for enrolled partition " + f"({marker.get('parent')!r} != {self.parent!r})" + ) + + def _enroll_new_parent(self) -> None: + marker = self._read_tail_marker() + if marker is not None: + if marker.get("parent") != self.parent: + raise BlockPoolError( + f"block pool parent {self.parent!r} has a devstack marker " + f"for a different path ({marker.get('parent')!r})" + ) + self._state["pool_id"] = marker.get("pool_id") + self._state["enrolled"] = True + return + + if not self.allow_enroll: + raise BlockPoolError( + f"block pool parent {self.parent!r} is not enrolled. " + "Use a dedicated empty partition and set " + "block_pool.allow_enroll = true for the first run only." + ) + if host.device_mounted(self.parent): + raise BlockPoolError( + f"block pool parent {self.parent!r} is mounted; refusing to enroll" + ) + if self._device_has_blkid_signature(): + raise BlockPoolError( + f"block pool parent {self.parent!r} has a filesystem or partition " + "table signature; refusing to enroll" + ) + + if self._has_partitions(os.path.basename(self.parent)): + raise BlockPoolError( + f"block pool parent {self.parent!r} is a whole disk with " + "existing partitions; use a dedicated empty disk or a single " + "partition instead" + ) + self._probe_parent_empty() + + payload = { + "pool_id": self._state["pool_id"], + "parent": self.parent, + } + self._write_tail_marker(payload) + self._state["enrolled"] = True + + def _validate_region(self, offset: int, size: int, owner: str) -> None: + parent_size = self._device_size + if offset + size > parent_size: + raise BlockPoolError( + f"block pool region for {owner!r} at offset {offset} with size " + f"{size} exceeds parent size {parent_size} for {self.parent!r}" + ) + if offset >= self._tail_marker_offset(): + raise BlockPoolError( + f"block pool region for {owner!r} would overlap the enrollment " + f"marker on {self.parent!r}" + ) + probe = min(_PROBE_SIZE, size) + for probe_offset in self._region_probe_offsets(offset, size): + data = self._read_parent(probe_offset, probe) + if not self._is_region_empty(data): + raise BlockPoolError( + f"block pool region for {owner!r} on {self.parent!r} is not " + f"empty at offset {probe_offset}; refusing to allocate over " + "existing data" + ) + + @staticmethod + def _region_probe_offsets(offset: int, size: int) -> list[int]: + """Parent offsets of the region's probe areas (see _BLUESTORE_LABEL_OFFSETS).""" + probe = min(_PROBE_SIZE, size) + return [ + offset + label + for label in _BLUESTORE_LABEL_OFFSETS + if label + probe <= size + ] + + def _pop_free_region(self, size: int) -> dict | None: + free_regions = self._state["free_regions"] + for index, region in enumerate(free_regions): + if region["size"] == size: + return free_regions.pop(index) + return None + + def get_or_allocate(self, owner: str, index: int, size: int) -> int: + """Offset of the region for owner's index-th device, allocating if new.""" + with self._state_lock(): + changed = self._ensure_ready_unlocked() + key = f"{owner}:{index}" + allocations = self._state["allocations"] + if key in allocations: + if changed: + self._atomic_save() + return allocations[key]["offset"] + + if region := self._pop_free_region(size): + offset = region["offset"] + else: + offset = self._state["next_offset"] + self._state["next_offset"] = offset + size + + self._validate_region(offset, size, owner) + allocations[key] = {"offset": offset, "size": size, "owner": owner} + self._atomic_save() + return offset + + def release_owner(self, owner: str) -> list[str]: + """Return owner's regions to the free list; list the ones kept back. + + Each region's probe areas are zeroed first, so that _validate_region + accepts it when it is handed out again, and no BlueStore label copy + left by its owner's OSD survives for ceph-volume to trip over; + anything else found there later is foreign data and still refused. A region that cannot be + zeroed stays allocated to owner (releasing again retries it), and + its ":" key is returned. Nothing is written, and every + region is kept, unless the parent still carries this pool's + enrollment marker: the device name may now point at another disk. + """ + kept = [] + with self._state_lock(): + prefix = f"{owner}:" + allocations = self._state["allocations"] + free_regions = self._state["free_regions"] + keys = [key for key in allocations if key.startswith(prefix)] + if not self.enrolled: + return keys + try: + self._verify_enrolled_parent() + except BlockPoolError: + return keys + for key in keys: + region = allocations[key] + probe = min(_PROBE_SIZE, region["size"]) + try: + for probe_offset in self._region_probe_offsets( + region["offset"], region["size"] + ): + self._write_parent(probe_offset, b"\x00" * probe) + except BlockPoolError: + kept.append(key) + continue + del allocations[key] + free_regions.append( + {"offset": region["offset"], "size": region["size"]} + ) + free_regions.sort(key=lambda region: region["offset"]) + self._atomic_save() + return kept + + +@functools.cache +def shared_pool() -> BlockPool | None: + """The block pool for the loaded config, built once per process and shared. + + State is re-read under the lock for every operation, so one instance can + serve every requirement check and provisioner in a run. + """ + return BlockPool.from_config(config) diff --git a/ceph_devstack/exec.py b/ceph_devstack/exec.py index 1bddb991..fded6a50 100644 --- a/ceph_devstack/exec.py +++ b/ceph_devstack/exec.py @@ -124,6 +124,13 @@ def run(self) -> subprocess.Popen: proc.wait() return proc + def capture(self, input: Optional[bytes] = None) -> subprocess.CompletedProcess: + """Run to completion, feeding input on stdin and capturing output.""" + logger.log(VERBOSE, self._make_log_msg()) + return subprocess.run( + self.args, env=self.env, input=input, capture_output=True, check=False + ) + async def arun(self) -> Subprocess: logger.log(VERBOSE, self._make_log_msg()) loop = asyncio.get_running_loop() diff --git a/ceph_devstack/host.py b/ceph_devstack/host.py index 8c744acc..8dae701b 100644 --- a/ceph_devstack/host.py +++ b/ceph_devstack/host.py @@ -1,14 +1,17 @@ +import functools import json import logging +import math import os import pathlib import shlex import socket +import subprocess import sys import yaml from packaging.version import parse as parse_version, Version -from typing import Dict, List, Optional, Union +from typing import Dict, List, Optional, Tuple, Union from .exec import Command, Subprocess @@ -56,6 +59,57 @@ def path_exists(self, path: Union[str, pathlib.Path]): path = pathlib.Path(path) return path.expanduser().exists() + def output( + self, args: List[str], input: Optional[bytes] = None + ) -> subprocess.CompletedProcess: + """Run args to completion on this host; capture stdout/stderr as bytes. + + This blocks the caller (and any running event loop) until the command + exits; it is meant for short queries and small device reads/writes. + """ + return self.cmd(args).capture(input) + + def read_text(self, path: Union[str, pathlib.Path]) -> Optional[str]: + """Contents of a file on this host, or None if it cannot be read.""" + proc = self.output(["cat", str(path)]) + return proc.stdout.decode() if proc.returncode == 0 else None + + def list_dir(self, path: Union[str, pathlib.Path]) -> List[str]: + """Entry names in a directory on this host; empty if it is unreadable.""" + proc = self.output(["ls", "-1", str(path)]) + return proc.stdout.decode().splitlines() if proc.returncode == 0 else [] + + def realpath(self, path: Union[str, pathlib.Path]) -> str: + """Canonical path on this host; the input unchanged if it can't resolve.""" + proc = self.output(["realpath", str(path)]) + return proc.stdout.decode().strip() if proc.returncode == 0 else str(path) + + def device_mounted(self, device: str) -> bool: + """True if a filesystem on this host is mounted from device.""" + proc = self.output(["findmnt", "-n", "-o", "TARGET", "--source", device]) + return proc.returncode == 0 and bool(proc.stdout.strip()) + + def read_bytes( + self, path: Union[str, pathlib.Path], offset: int, length: int + ) -> bytes: + """Read length (> 0) bytes at offset from a file or device on this host.""" + bs = math.gcd(offset, length) + args = ["dd", f"if={path}", f"bs={bs}", f"skip={offset // bs}"] + proc = self.output(args + [f"count={length // bs}"]) + if proc.returncode != 0: + raise OSError(f"cannot read {path}: {proc.stderr.decode().strip()}") + return proc.stdout + + def write_bytes( + self, path: Union[str, pathlib.Path], offset: int, data: bytes + ) -> None: + """Write data (non-empty) at offset into a file or device on this host.""" + bs = math.gcd(offset, len(data)) + args = ["dd", f"of={path}", f"bs={bs}", f"seek={offset // bs}"] + proc = self.output(args + ["conv=notrunc,fsync"], input=data) + if proc.returncode != 0: + raise OSError(f"cannot write {path}: {proc.stderr.decode().strip()}") + def hostname(self) -> str: name = socket.getfqdn() try: @@ -138,6 +192,10 @@ async def apparmor_enabled(self) -> bool: return False return await proc.wait() == 0 + def user_ids(self) -> Tuple[int, int]: + """Return (uid, gid) of the user running this host.""" + return (os.getuid(), os.getgid()) + class LocalHost(Host): pass @@ -148,9 +206,9 @@ class RemoteHost(Host): base_args = ["podman", "machine", "ssh"] def _remote_args(self, args: List[str], stream_output: bool) -> List[str]: + # Quote each arg so compound arguments like "bash -c ''" stay as one shell word remote = list(self.base_args) remote.append("--") - # Quote each arg so compound arguments like "bash -c ''" stay as one shell word remote.extend(shlex.quote(arg) for arg in args) return remote @@ -175,6 +233,21 @@ def hostname(self) -> str: assert proc.stdout is not None return proc.stdout.read().decode().strip() + def user_ids(self) -> Tuple[int, int]: + """Return (uid, gid) of the user running podman on the container host.""" + return self._user_ids + + @functools.cached_property + def _user_ids(self) -> Tuple[int, int]: + return self._id("-u"), self._id("-g") + + def _id(self, flag: str) -> int: + out = self.output(["id", flag]).stdout.decode().strip() + try: + return int(out) + except ValueError as e: + raise RuntimeError(f"Failed to parse output of 'id {flag}': {out!r}") from e + local_host = LocalHost() diff --git a/ceph_devstack/resources/ceph/block_devices.py b/ceph_devstack/resources/ceph/block_devices.py new file mode 100644 index 00000000..9cf2e3ef --- /dev/null +++ b/ceph_devstack/resources/ceph/block_devices.py @@ -0,0 +1,254 @@ +import asyncio +import json +from pathlib import Path +from typing import Protocol + +from ceph_devstack import logger +from ceph_devstack.block_pool import BlockPool, BlockPoolError, format_size, parse_size +from ceph_devstack.exec import Subprocess +from ceph_devstack.host import host + +# After detaching, wait up to _DETACH_CHECKS * _DETACH_RETRY_DELAY seconds for +# busy loops (e.g. still held by udev) to let go of their pool regions. +_DETACH_CHECKS = 5 +_DETACH_RETRY_DELAY = 0.5 + + +def backing_name(owner: str, loop_number: int) -> str: + """Backing file name for a loop device owned by a container.""" + return f"{owner}-{loop_number}" + + +def parse_backing_name(name: str) -> tuple[str, int] | None: + """Parse backing file name into (owner, loop_number) or return None.""" + owner, sep, loop_id = name.rpartition("-") + if not sep or not loop_id.isdigit(): + return None + return owner, int(loop_id) + + +def loop_path(number: int) -> str: + """Host /dev/loop path for a loop number.""" + return f"/dev/loop{number}" + + +def loop_number(path: str) -> int: + """Extract loop number from /dev/loop path.""" + return int(path.removeprefix("/dev/loop")) + + +def attached_loops() -> list[tuple[int, int, str]]: + """(loop number, offset, backing file) of every attached loop on the host. + + One ``losetup --list`` call reads all of it from sysfs; no root needed. + Raises BlockPoolError if the loops can't be listed: callers decide what is + free or safe to release from this, so a failure must never look empty. + """ + proc = host.output( + ["losetup", "--list", "--json", "--output", "NAME,OFFSET,BACK-FILE"] + ) + if proc.returncode != 0: + raise BlockPoolError( + f"cannot list loop devices: {proc.stderr.decode().strip()}" + ) + if not proc.stdout.strip(): + return [] # losetup prints nothing when no loops are attached + try: + return [ + (loop_number(loop["name"]), int(loop["offset"]), loop["back-file"]) + for loop in json.loads(proc.stdout)["loopdevices"] + ] + except (ValueError, KeyError, TypeError) as e: + raise BlockPoolError(f"cannot parse losetup output: {e}") from e + + +def backing_files(image_dir: Path) -> list[tuple[str, int]]: + """(owner, loop number) for each backing file in image_dir on the host.""" + return [ + parsed + for name in host.list_dir(image_dir) + if (parsed := parse_backing_name(name)) is not None + ] + + +class CommandRunner(Protocol): + async def __call__( + self, + args: list[str], + *, + check: bool = False, + ) -> Subprocess: ... + + +class BlockDeviceProvisioner: + """Create loop devices backed by sparse files or NVMe pool slices. + + All commands, including backing-file management in image_dir, run on the + container host: losetup resolves image paths there. + """ + + def __init__( + self, + owner: str, + *, + image_dir: Path, + file_size: str, + cmd: CommandRunner, + trigger_udev: bool = False, + pool: BlockPool | None, + ): + self.owner = owner + self.image_dir = Path(image_dir).expanduser().absolute() + self.device_size = parse_size(file_size) + self.trigger_udev = trigger_udev + self._cmd = cmd + self.pool = pool + + def _image_path(self, device: str) -> Path: + num = loop_number(device) + return self.image_dir / backing_name(self.owner, num) + + async def create_devices(self, devices: list[str]) -> None: + for index, device in enumerate(devices): + await self.create_device(device, index) + + def owned_devices(self) -> list[str]: + """Loop devices this owner holds on the host, found from host state. + + Pool mode: loops attached to the pool parent at one of the owner's + region offsets. Sparse mode: loops named by the owner's backing files. + """ + if (pool := self.pool) is not None: + offsets = pool.owner_offsets(self.owner) + numbers = [ + number + for number, offset, backing in attached_loops() + if backing == pool.parent and offset in offsets + ] + else: + numbers = [ + number + for owner, number in backing_files(self.image_dir) + if owner == self.owner + ] + return [loop_path(number) for number in sorted(numbers)] + + async def remove_devices(self, devices: list[str]) -> None: + """Detach devices; then release the owner's pool regions once detached. + + If loops stay attached (e.g. busy), or a region's start can't be + zeroed, the regions stay allocated and a warning says so; removing + again later releases them. + """ + for device in devices: + await self.remove_device(device) + if (pool := self.pool) is None: + return + await self._cmd(["sudo", "udevadm", "settle"], check=False) + for attempt in range(_DETACH_CHECKS): + if not (still_attached := self.owned_devices()): + if kept := pool.release_owner(self.owner): + logger.warning( + f"{self.owner}: could not wipe pool regions " + f"{', '.join(kept)} on {pool.parent}; keeping them " + "allocated until a later remove succeeds" + ) + return + if attempt < _DETACH_CHECKS - 1: + await asyncio.sleep(_DETACH_RETRY_DELAY) + logger.warning( + f"{self.owner}: {', '.join(still_attached)} still attached to " + f"{pool.parent}; keeping its pool regions allocated until a later " + "remove succeeds" + ) + + async def create_device(self, device: str, index: int) -> None: + await self._ensure_loop_module() + await self.remove_device(device) + try: + await self._make_device(device, index) + except Exception: + # Removal finds devices from attached loops (pool) or backing + # files (sparse), so a node left behind here would never be found. + try: + await self.remove_device(device) + except Exception: + logger.exception(f"{self.owner}: could not clean up {device}") + raise + + async def _make_device(self, device: str, index: int) -> None: + """Create the device node and attach it to its pool region or file.""" + device_pos = str(loop_number(device)) + await self._cmd( + ["sudo", "mknod", "-m700", device, "b", "7", device_pos], + check=True, + ) + # Use host.user_ids() to get the container host's uid:gid (not the local process uid:gid) + uid, gid = host.user_ids() + await self._cmd( + ["sudo", "chown", f"{uid}:{gid}", device], + check=True, + ) + if (pool := self.pool) is not None: + offset = pool.get_or_allocate(self.owner, index, self.device_size) + logger.info( + f"{self.owner}: region on {pool.parent} " + f"(offset={offset}, size={format_size(self.device_size)}) -> {device}" + ) + await self._cmd( + [ + "sudo", + "losetup", + "--offset", + str(offset), + "--sizelimit", + str(self.device_size), + device, + pool.parent, + ], + check=True, + ) + else: + image_path = self._image_path(device) + await self._cmd(["mkdir", "-p", str(self.image_dir)], check=True) + await self._cmd( + [ + "sudo", + "dd", + "if=/dev/null", + f"of={image_path}", + "bs=1", + "count=0", + f"seek={self.device_size}", + ], + check=True, + ) + await self._cmd(["sudo", "losetup", device, str(image_path)], check=True) + + await self._cmd(["chcon", "-t", "fixed_disk_device_t", device], check=False) + if self.trigger_udev: + await self._cmd( + [ + "sudo", + "udevadm", + "trigger", + "--action=add", + f"--name=block/loop{device_pos}", + ], + check=False, + ) + await self._cmd(["sudo", "udevadm", "settle"], check=False) + + async def remove_device(self, device: str) -> None: + if host.device_mounted(device): + await self._cmd(["umount", device], check=True) + if host.path_exists(device): + await self._cmd(["sudo", "losetup", "-d", device], check=False) + await self._cmd(["sudo", "rm", "-f", device], check=False) + if self.pool is None: + await self._cmd(["rm", "-f", str(self._image_path(device))], check=False) + + async def _ensure_loop_module(self) -> None: + proc = await self._cmd(["bash", "-c", "lsmod | grep -q loop"], check=False) + if await proc.wait() != 0: + await self._cmd(["sudo", "modprobe", "loop"]) diff --git a/ceph_devstack/resources/ceph/host_loops.py b/ceph_devstack/resources/ceph/host_loops.py new file mode 100644 index 00000000..51f1e4b7 --- /dev/null +++ b/ceph_devstack/resources/ceph/host_loops.py @@ -0,0 +1,123 @@ +"""Host loop device discovery, allocation, and provisioning mixin.""" + +from pathlib import Path +from typing import Any, ClassVar + +from ceph_devstack import logger +from ceph_devstack.block_pool import shared_pool +from ceph_devstack.resources.ceph.block_devices import ( + BlockDeviceProvisioner, + attached_loops, + backing_files, + loop_number, + loop_path, +) + + +class LoopDeviceMixin: + """Mixin for Container subclasses that manage host loop devices. + + Subclasses must define a ``loop_img_dir`` property returning a ``Path``. + Set ``trigger_udev = True`` on the class to trigger udev after device creation. + """ + + name: str + config: dict[str, Any] + cmd: Any + trigger_udev: bool = False + _allocated: ClassVar[set[int]] = set() + + @property + def loop_img_dir(self) -> Path: + raise NotImplementedError + + @classmethod + def discover_used_loop_numbers(cls, image_dir: Path) -> set[int]: + """Loop numbers already attached on the host or claimed by backing files.""" + attached = {number for number, _, _ in attached_loops()} + return attached | {number for _, number in backing_files(image_dir)} + + @classmethod + def owner_loop_numbers(cls, owner: str, image_dir: Path) -> list[int]: + """Existing backing-file loop numbers for an owner, lowest first.""" + return sorted(n for o, n in backing_files(image_dir) if o == owner) + + @classmethod + def allocate_loop_numbers( + cls, owner: str, count: int, image_dir: Path + ) -> list[int]: + """Pick loop numbers for an owner from config count and host availability.""" + used = cls.discover_used_loop_numbers(image_dir) | cls._allocated + reclaimed = cls.owner_loop_numbers(owner, image_dir) + numbers: list[int] = [] + for index in range(count): + if index < len(reclaimed): + numbers.append(reclaimed[index]) + continue + candidate = 0 + while candidate in used or candidate in numbers: + candidate += 1 + numbers.append(candidate) + used.add(candidate) + cls._allocated.update(numbers) + return numbers + + @classmethod + def allocate_loop_devices( + cls, owner: str, count: int, image_dir: Path + ) -> list[str]: + return [ + loop_path(number) + for number in cls.allocate_loop_numbers(owner, count, image_dir) + ] + + def _init_loop_devices(self) -> None: + self.loop_device_count: int = self.config["loop_device_count"] + self._devices: list[str] | None = None + self._block_device_provisioner: BlockDeviceProvisioner | None = None + + def allocate_devices(self) -> list[str]: + """Pick this container's host loop devices now, if not chosen already.""" + if self._devices is None: + self._devices = self.allocate_loop_devices( + self.name, self.loop_device_count, self.loop_img_dir + ) + return self._devices + + @property + def devices(self) -> list[str]: + return self.allocate_devices() + + def _block_provisioner(self) -> BlockDeviceProvisioner: + if self._block_device_provisioner is None: + self._block_device_provisioner = BlockDeviceProvisioner( + self.name, + image_dir=self.loop_img_dir, + file_size=self.config["loop_device_size"], + cmd=self.cmd, + trigger_udev=self.trigger_udev, + pool=shared_pool(), + ) + return self._block_device_provisioner + + async def create_loop_devices(self): + if self.devices: + numbers = ", ".join(str(loop_number(d)) for d in self.devices) + logger.info( + f"{self.name}: host loop devices {numbers} " + f"({self.config['loop_device_size']} each)" + ) + await self._block_provisioner().create_devices(self.devices) + + async def remove_loop_devices(self): + """Detach the devices this container holds on the host. + + Devices are found from host state (plus any picked earlier in this + process), so removal never allocates fresh numbers that might belong + to someone else. Their numbers are then released for reallocation. + """ + provisioner = self._block_provisioner() + devices = set(provisioner.owned_devices()) | set(self._devices or []) + await provisioner.remove_devices(sorted(devices, key=loop_number)) + self._allocated.difference_update(loop_number(d) for d in devices) + self._devices = None diff --git a/tests/conftest.py b/tests/conftest.py index 6dfea423..40fc5b91 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,19 +1,165 @@ +import json import os import pathlib -import pytest import random - +import subprocess +from collections.abc import Callable from datetime import datetime, timedelta +from unittest.mock import AsyncMock, patch + +import pytest from ceph_devstack import config +from ceph_devstack.block_pool import BlockPool, shared_pool +from ceph_devstack.host import Host +from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin @pytest.fixture(autouse=True) def reset_config(): config.load() + shared_pool.cache_clear() + LoopDeviceMixin._allocated.clear() yield +class FakeHost(Host): + """In-memory host for block-pool and loop-device commands. + + ``files`` maps absolute paths to contents; directories are implied by path + prefixes. ``loops`` lists attached loop devices as (number, offset, + backing file) for ``losetup --list``. ``responses`` gives (returncode, + stdout) by argv[0] and overrides everything else; the defaults describe an + unmounted device without signatures. Any other command fails the test. + Every command run through ``output`` is recorded in ``commands``. + """ + + def __init__(self): + self.files: dict[str, bytearray] = {} + self.loops: list[tuple[int, int, str]] = [] + self.responses: dict[str, tuple[int, bytes]] = { + "findmnt": (1, b""), + "blkid": (2, b""), + } + self.commands: list[list[str]] = [] + + def add_file(self, path: str | pathlib.Path, data: bytes | str = b"") -> None: + if isinstance(data, str): + data = data.encode() + self.files[str(path)] = bytearray(data) + + def add_block_device( + self, name: str, size: int = 20 * 1024**2, disk: str | None = None + ) -> str: + """Add /dev/ of size zero bytes and its sysfs entry; return the path. + + For a partition, ``disk`` also lists it under the disk's sysfs directory. + """ + self.add_file(f"/dev/{name}", bytes(size)) + self.add_file(f"/sys/class/block/{name}/size", f"{size // 512}\n") + if disk: + self.add_file(f"/sys/class/block/{disk}/{name}/size", f"{size // 512}\n") + return f"/dev/{name}" + + def _children(self, path: str) -> list[str]: + prefix = path.rstrip("/") + "/" + return sorted( + {p[len(prefix) :].split("/")[0] for p in self.files if p.startswith(prefix)} + ) + + def path_exists(self, path) -> bool: + return str(path) in self.files or bool(self._children(str(path))) + + def output(self, args, input=None) -> subprocess.CompletedProcess: + self.commands.append(args) + name, *rest = args + if name in self.responses: + rc, out = self.responses[name] + elif name == "cat": + data = self.files.get(rest[0]) + rc, out = (1, b"") if data is None else (0, bytes(data)) + elif name == "ls": + children = self._children(rest[-1]) + rc, out = (0, "".join(f"{c}\n" for c in children).encode()) + elif name == "realpath": + rc, out = 0, f"{rest[0]}\n".encode() + elif name == "losetup" and rest[0] == "--list": + loops = [ + {"name": f"/dev/loop{n}", "offset": off, "back-file": back} + for n, off, back in self.loops + ] + rc, out = 0, (json.dumps({"loopdevices": loops}).encode() if loops else b"") + elif name == "dd": + opts = dict(arg.split("=", 1) for arg in rest) + bs = int(opts["bs"]) + rc, out = 0, b"" + if "if" in opts: + start = int(opts["skip"]) * bs + out = bytes( + self.files[opts["if"]][start : start + int(opts["count"]) * bs] + ) + else: + assert input is not None and opts.get("conv") == "notrunc,fsync" + start = int(opts["seek"]) * bs + self.files[opts["of"]][start : start + len(input)] = input + else: + raise AssertionError(f"unexpected host command {args}") + return subprocess.CompletedProcess(args, rc, out, b"") + + +@pytest.fixture(autouse=True) +def fake_host(): + """Isolate block-pool and loop-device code from the real host. + + Nothing reads the real /sys/class/block or block devices; tests that need + sysfs entries, devices, or backing files add them to the fake. + """ + fake = FakeHost() + with ( + patch("ceph_devstack.block_pool.host", fake), + patch("ceph_devstack.resources.ceph.block_devices.host", fake), + ): + yield fake + + +@pytest.fixture +def mock_cmd() -> AsyncMock: + return AsyncMock() + + +@pytest.fixture +def pool_parent(fake_host: FakeHost) -> str: + return fake_host.add_block_device("nvme0n1p1") + + +@pytest.fixture +def block_pool_factory( + tmp_path: pathlib.Path, pool_parent: str +) -> Callable[..., BlockPool]: + def factory(*, allow_enroll: bool = True) -> BlockPool: + return BlockPool( + tmp_path / "block_pool.json", pool_parent, allow_enroll=allow_enroll + ) + + return factory + + +@pytest.fixture +def block_pool(block_pool_factory: Callable[..., BlockPool]) -> BlockPool: + return block_pool_factory() + + +@pytest.fixture +def enrolled_pool(block_pool: BlockPool) -> BlockPool: + block_pool.ensure_ready() + return block_pool + + +@pytest.fixture +def disallow_enroll_pool(block_pool_factory: Callable[..., BlockPool]) -> BlockPool: + return block_pool_factory(allow_enroll=False) + + @pytest.fixture(scope="function") def create_log_file(): def _create_log_file(data_dir: pathlib.Path, **kwargs) -> pathlib.Path: diff --git a/tests/resources/ceph/conftest.py b/tests/resources/ceph/conftest.py new file mode 100644 index 00000000..d25b9da4 --- /dev/null +++ b/tests/resources/ceph/conftest.py @@ -0,0 +1,50 @@ +from collections.abc import Iterator +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from ceph_devstack.resources.ceph.block_devices import BlockDeviceProvisioner + + +@pytest.fixture +def pool_provisioner_factory(tmp_path: Path, block_pool, mock_cmd: AsyncMock): + def factory(file_size: str = "1M") -> BlockDeviceProvisioner: + return BlockDeviceProvisioner( + "ceph_node", + image_dir=tmp_path / "disk_images", + file_size=file_size, + cmd=mock_cmd, + pool=block_pool, + ) + + return factory + + +@pytest.fixture +def pool_provisioner(pool_provisioner_factory) -> BlockDeviceProvisioner: + return pool_provisioner_factory() + + +@pytest.fixture +def ready_pool_provisioner( + pool_provisioner: BlockDeviceProvisioner, +) -> Iterator[BlockDeviceProvisioner]: + with patch.object(pool_provisioner, "remove_device", new=AsyncMock()): + yield pool_provisioner + + +@pytest.fixture +def sparse_provisioner(tmp_path: Path, mock_cmd: AsyncMock) -> BlockDeviceProvisioner: + return BlockDeviceProvisioner( + "testnode_0", + image_dir=tmp_path / "disk_images", + file_size="1M", + cmd=mock_cmd, + pool=None, + ) + + +@pytest.fixture +def large_pool_provisioner(pool_provisioner_factory) -> BlockDeviceProvisioner: + return pool_provisioner_factory(file_size="32M") diff --git a/tests/resources/ceph/test_block_devices.py b/tests/resources/ceph/test_block_devices.py new file mode 100644 index 00000000..992602f7 --- /dev/null +++ b/tests/resources/ceph/test_block_devices.py @@ -0,0 +1,226 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from ceph_devstack.block_pool import BlockPoolError +from ceph_devstack.resources.ceph.block_devices import ( + BlockDeviceProvisioner, + attached_loops, +) + +MiB = 1024**2 + + +class TestBlockDeviceProvisioner: + async def test_create_device_uses_pool_slice( + self, + ready_pool_provisioner: BlockDeviceProvisioner, + mock_cmd: AsyncMock, + pool_parent, + ): + await ready_pool_provisioner.create_device("/dev/loop9", 0) + losetup = [ + call.args[0] + for call in mock_cmd.await_args_list + if "losetup" in call.args[0] + ][0] + offset_idx = losetup.index("--offset") + assert losetup[offset_idx + 1] == "0" + sizelimit_idx = losetup.index("--sizelimit") + assert losetup[sizelimit_idx + 1] == str(ready_pool_provisioner.device_size) + assert losetup[-2:] == ["/dev/loop9", pool_parent] + + async def test_remove_devices_releases_pool_owner( + self, + ready_pool_provisioner: BlockDeviceProvisioner, + ): + await ready_pool_provisioner.create_device("/dev/loop9", 0) + pool = ready_pool_provisioner.pool + assert pool is not None + assert list(pool.allocations) == ["ceph_node:0"] + await ready_pool_provisioner.remove_devices(["/dev/loop9"]) + assert pool.allocations == {} + assert pool.free_regions == [{"offset": 0, "size": 1024**2}] + + async def test_failed_create_removes_the_device_node( + self, + pool_provisioner: BlockDeviceProvisioner, + mock_cmd: AsyncMock, + ): + pool = pool_provisioner.pool + assert pool is not None + with ( + patch.object(pool_provisioner, "remove_device", new=AsyncMock()) as remove, + patch.object( + pool, "get_or_allocate", side_effect=BlockPoolError("not empty") + ), + pytest.raises(BlockPoolError), + ): + await pool_provisioner.create_device("/dev/loop9", 0) + assert ["sudo", "mknod", "-m700", "/dev/loop9", "b", "7", "9"] in [ + call.args[0] for call in mock_cmd.await_args_list + ] + assert remove.await_args_list == [(("/dev/loop9",),)] * 2 + + async def test_failed_cleanup_does_not_hide_the_create_error( + self, + pool_provisioner: BlockDeviceProvisioner, + caplog, + ): + pool = pool_provisioner.pool + assert pool is not None + cleanup = AsyncMock(side_effect=[None, RuntimeError("umount failed")]) + with ( + patch.object(pool_provisioner, "remove_device", new=cleanup), + patch.object( + pool, "get_or_allocate", side_effect=BlockPoolError("not empty") + ), + pytest.raises(BlockPoolError, match="not empty"), + ): + await pool_provisioner.create_device("/dev/loop9", 0) + assert "could not clean up /dev/loop9" in caplog.text + + async def test_remove_devices_warns_about_regions_it_cannot_wipe( + self, + ready_pool_provisioner: BlockDeviceProvisioner, + caplog, + ): + pool = ready_pool_provisioner.pool + assert pool is not None + with patch.object(pool, "release_owner", return_value=["ceph_node:0"]): + await ready_pool_provisioner.remove_devices([]) + assert "could not wipe pool regions ceph_node:0" in caplog.text + + async def test_create_device_uses_sparse_file_without_pool( + self, + sparse_provisioner: BlockDeviceProvisioner, + mock_cmd: AsyncMock, + tmp_path, + ): + with ( + patch.object(sparse_provisioner, "remove_device", new=AsyncMock()), + patch.object(sparse_provisioner, "_ensure_loop_module", new=AsyncMock()), + ): + await sparse_provisioner.create_device("/dev/loop1", 0) + image_dir = tmp_path / "disk_images" + image_path = str(image_dir / "testnode_0-1") + commands = [call.args[0] for call in mock_cmd.await_args_list] + assert ["mkdir", "-p", str(image_dir)] in commands + dd = next(c for c in commands if "dd" in c) + assert f"of={image_path}" in dd + assert f"seek={sparse_provisioner.device_size}" in dd + assert ["sudo", "losetup", "/dev/loop1", image_path] in commands + + async def test_remove_device_deletes_backing_file_on_host( + self, + sparse_provisioner: BlockDeviceProvisioner, + mock_cmd: AsyncMock, + tmp_path, + ): + await sparse_provisioner.remove_device("/dev/loop1") + image_path = str(tmp_path / "disk_images" / "testnode_0-1") + mock_cmd.assert_awaited_once_with(["rm", "-f", image_path], check=False) + + async def test_rejects_region_larger_than_parent( + self, + large_pool_provisioner: BlockDeviceProvisioner, + ): + with ( + patch.object(large_pool_provisioner, "remove_device", new=AsyncMock()), + pytest.raises(BlockPoolError, match="exceeds"), + ): + await large_pool_provisioner.create_device("/dev/loop9", 0) + + async def test_ensure_loop_module_loads_when_missing( + self, + sparse_provisioner: BlockDeviceProvisioner, + mock_cmd: AsyncMock, + ): + mock_proc = AsyncMock() + mock_proc.wait = AsyncMock(return_value=1) + mock_cmd.return_value = mock_proc + await sparse_provisioner._ensure_loop_module() + mock_cmd.assert_any_await(["sudo", "modprobe", "loop"]) + + +class TestHostLoopDiscovery: + def test_attached_loops_reads_losetup_list(self, fake_host): + fake_host.responses["losetup"] = ( + 0, + b'{"loopdevices": [{"name": "/dev/loop7", "offset": "4096",' + b' "back-file": "/dev/sda1"}]}', + ) + assert attached_loops() == [(7, 4096, "/dev/sda1")] + assert fake_host.commands == [ + ["losetup", "--list", "--json", "--output", "NAME,OFFSET,BACK-FILE"] + ] + + def test_pool_owner_devices_are_its_loops_on_its_regions( + self, fake_host, pool_provisioner + ): + pool = pool_provisioner.pool + assert pool is not None + mine = [pool.get_or_allocate("ceph_node", i, MiB) for i in (0, 1)] + other = pool.get_or_allocate("testnode_0", 0, MiB) + fake_host.loops = [ + (9, mine[1], pool.parent), + (7, mine[0], pool.parent), + (3, other, pool.parent), + (4, mine[0], "/var/lib/other.img"), + ] + assert pool_provisioner.owned_devices() == ["/dev/loop7", "/dev/loop9"] + + def test_sparse_owner_devices_come_from_backing_files( + self, fake_host, sparse_provisioner, tmp_path + ): + for name in ("testnode_0-5", "testnode_0-2", "testnode_01-3"): + fake_host.add_file(tmp_path / "disk_images" / name) + assert sparse_provisioner.owned_devices() == ["/dev/loop2", "/dev/loop5"] + + async def test_regions_stay_allocated_while_loops_remain_attached( + self, fake_host, ready_pool_provisioner, mock_cmd, caplog + ): + pool = ready_pool_provisioner.pool + assert pool is not None + await ready_pool_provisioner.create_device("/dev/loop9", 0) + fake_host.loops = [(9, 0, pool.parent)] # detach did not take effect + with patch("ceph_devstack.resources.ceph.block_devices.asyncio.sleep") as sleep: + await ready_pool_provisioner.remove_devices(["/dev/loop9"]) + mock_cmd.assert_any_await(["sudo", "udevadm", "settle"], check=False) + assert sleep.await_count == 4 + assert "/dev/loop9 still attached" in caplog.text + assert list(pool.allocations) == ["ceph_node:0"] + + async def test_regions_released_once_busy_loop_lets_go( + self, fake_host, ready_pool_provisioner + ): + pool = ready_pool_provisioner.pool + assert pool is not None + await ready_pool_provisioner.create_device("/dev/loop9", 0) + fake_host.loops = [(9, 0, pool.parent)] + + async def detach_later(delay): + fake_host.loops = [] + + with patch( + "ceph_devstack.resources.ceph.block_devices.asyncio.sleep", + side_effect=detach_later, + ): + await ready_pool_provisioner.remove_devices(["/dev/loop9"]) + assert pool.allocations == {} + + @pytest.mark.parametrize( + "response", + [(1, b""), (0, b"not json"), (0, b'{"other": []}')], + ids=["losetup-fails", "garbage", "unexpected-json"], + ) + async def test_failed_loop_discovery_aborts_removal( + self, fake_host, ready_pool_provisioner, response + ): + pool = ready_pool_provisioner.pool + assert pool is not None + await ready_pool_provisioner.create_device("/dev/loop9", 0) + fake_host.responses["losetup"] = response + with pytest.raises(BlockPoolError, match="losetup|loop devices"): + await ready_pool_provisioner.remove_devices(["/dev/loop9"]) + assert list(pool.allocations) == ["ceph_node:0"] diff --git a/tests/resources/ceph/test_host_loops.py b/tests/resources/ceph/test_host_loops.py new file mode 100644 index 00000000..d2b46966 --- /dev/null +++ b/tests/resources/ceph/test_host_loops.py @@ -0,0 +1,117 @@ +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from ceph_devstack.block_pool import BlockPoolError +from ceph_devstack.resources.ceph.block_devices import ( + BlockDeviceProvisioner, + loop_number, + loop_path, +) +from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin + + +def make_backing(fake_host, image_dir, *names): + """Create empty backing files for loop devices on the (fake) host.""" + for name in names: + fake_host.add_file(image_dir / name) + + +def make_node(name, image_dir, count=2): + class Node(LoopDeviceMixin): + loop_img_dir = image_dir + cmd = AsyncMock() + + def __init__(self): + self.name = name + self.config = {"loop_device_count": count, "loop_device_size": "1M"} + self._init_loop_devices() + + return Node() + + +class TestHostLoops: + def test_loop_path_round_trips(self): + assert loop_path(21) == "/dev/loop21" + assert loop_number(loop_path(21)) == 21 + + def test_discover_used_loop_numbers_from_backing_files(self, fake_host, tmp_path): + make_backing(fake_host, tmp_path, "testnode_0-1", "ceph_node-4", "ignore-me") + assert LoopDeviceMixin.discover_used_loop_numbers(tmp_path) == {1, 4} + + def test_discover_used_loop_numbers_includes_attached_loops( + self, fake_host, tmp_path + ): + fake_host.loops = [(0, 0, "/img"), (5, 1048576, "/dev/nvme0n1p1")] + make_backing(fake_host, tmp_path, "testnode_0-3") + assert LoopDeviceMixin.discover_used_loop_numbers(tmp_path) == {0, 3, 5} + + def test_allocation_fails_when_attached_loops_are_unknown( + self, fake_host, tmp_path + ): + fake_host.responses["losetup"] = (1, b"") + with pytest.raises(BlockPoolError, match="cannot list loop devices"): + LoopDeviceMixin.allocate_loop_numbers("testnode_0", 1, tmp_path) + + def test_owner_loop_numbers(self, fake_host, tmp_path): + make_backing( + fake_host, tmp_path, "testnode_1-2", "testnode_1-0", "testnode_10-1" + ) + assert LoopDeviceMixin.owner_loop_numbers("testnode_1", tmp_path) == [0, 2] + + def test_allocate_loop_numbers_uses_lowest_free(self, fake_host, tmp_path): + make_backing(fake_host, tmp_path, "other-0") + assert LoopDeviceMixin.allocate_loop_numbers("testnode_0", 2, tmp_path) == [ + 1, + 2, + ] + + def test_allocate_loop_numbers_reuses_owner_backing_files( + self, fake_host, tmp_path + ): + make_backing(fake_host, tmp_path, "testnode_1-4", "testnode_1-1", "other-0") + assert LoopDeviceMixin.allocate_loop_numbers("testnode_1", 3, tmp_path) == [ + 1, + 4, + 2, + ] + + async def test_remove_targets_held_devices_without_allocating( + self, fake_host, tmp_path + ): + """A fresh process (e.g. `remove`) acts on what the host shows it holds.""" + make_backing(fake_host, tmp_path, "testnode_0-4", "other-0") + node = make_node("testnode_0", tmp_path) + with ( + patch.object(LoopDeviceMixin, "allocate_loop_devices") as allocate, + patch.object(BlockDeviceProvisioner, "remove_devices") as remove, + ): + await node.remove_loop_devices() + allocate.assert_not_called() + remove.assert_awaited_once_with(["/dev/loop4"]) + + def test_allocate_loop_numbers_avoids_cross_call_collisions( + self, fake_host, tmp_path + ): + """Multiple owners allocating from the same dir get distinct numbers.""" + make_backing(fake_host, tmp_path, "existing-0") + a = LoopDeviceMixin.allocate_loop_numbers("testnode_0", 1, tmp_path) + b = LoopDeviceMixin.allocate_loop_numbers("testnode_1", 1, tmp_path) + c = LoopDeviceMixin.allocate_loop_numbers("testnode_2", 1, tmp_path) + assert a == [1] + assert b == [2] + assert c == [3] + + async def test_removed_devices_are_released_for_reallocation(self, tmp_path): + first = make_node("testnode_0", tmp_path) + assert first.allocate_devices() == ["/dev/loop0", "/dev/loop1"] + provisioner = MagicMock( + owned_devices=MagicMock(return_value=[]), remove_devices=AsyncMock() + ) + with patch.object(first, "_block_provisioner", return_value=provisioner): + await first.remove_loop_devices() + provisioner.remove_devices.assert_awaited_once_with( + ["/dev/loop0", "/dev/loop1"] + ) + second = make_node("testnode_1", tmp_path) + assert second.allocate_devices() == ["/dev/loop0", "/dev/loop1"] diff --git a/tests/test_block_pool.py b/tests/test_block_pool.py new file mode 100644 index 00000000..b7e35331 --- /dev/null +++ b/tests/test_block_pool.py @@ -0,0 +1,330 @@ +import json +from unittest.mock import patch + +import pytest + +from ceph_devstack import config +from ceph_devstack.block_pool import ( + BlockPool, + BlockPoolError, + format_size, + parse_size, + shared_pool, + validate_parent_name, +) + +MiB = 1024**2 +TAIL = 20 * MiB - 4096 # tail marker offset on the default 20 MiB fake parent + + +def write(fake_host, path: str, offset: int, data: bytes) -> None: + fake_host.files[path][offset : offset + len(data)] = data + + +class TestParseSize: + def test_gigabytes(self): + assert parse_size("50G") == 50 * 1024**3 + + def test_integer(self): + assert parse_size(1024) == 1024 + + @pytest.mark.parametrize("value", ["not-a-size", "0", 0, "0G"]) + def test_invalid(self, value): + with pytest.raises(BlockPoolError): + parse_size(value) + + def test_format_size_human_units(self): + assert format_size(5 * 1024**3) == "5G" + assert format_size(12345) == "12345B" + + +class TestValidateParentName: + @pytest.mark.parametrize("name", ["nvme0n1p1", "nvme0n1", "sda1"]) + def test_accepts_disks_and_partitions(self, fake_host, name): + parent = fake_host.add_block_device(name) + assert validate_parent_name(parent) == parent + + def test_resolves_symlinks_on_host(self, fake_host): + fake_host.add_block_device("nvme0n1p1") + fake_host.responses["realpath"] = (0, b"/dev/nvme0n1p1\n") + assert validate_parent_name("/dev/disk/by-id/x") == "/dev/nvme0n1p1" + assert ["realpath", "/dev/disk/by-id/x"] in fake_host.commands + + def test_rejects_device_mapper(self): + with pytest.raises(BlockPoolError, match="not allowed"): + validate_parent_name("/dev/dm-0") + + def test_rejects_non_dev_path(self): + with pytest.raises(BlockPoolError, match="must be a /dev path"): + validate_parent_name("/tmp/not-a-device") + + def test_rejects_missing_device(self): + with pytest.raises(BlockPoolError, match="does not exist"): + validate_parent_name("/dev/nvme0n1p1") + + def test_rejects_non_block_device(self, fake_host): + fake_host.add_file("/dev/nvme0n1p1") + with pytest.raises(BlockPoolError, match="not a block device"): + validate_parent_name("/dev/nvme0n1p1") + + def test_rejects_unknown_block_name(self, fake_host): + fake_host.add_file("/dev/foo") + with pytest.raises(BlockPoolError, match="block device or partition"): + validate_parent_name("/dev/foo") + + +class TestBlockPool: + def test_allocate_and_reuse_stable_owner(self, enrolled_pool): + first_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + second_offset = enrolled_pool.get_or_allocate("ceph_node", 1, MiB) + again_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + assert (first_offset, second_offset, again_offset) == (0, MiB, 0) + + def test_release_and_reuse_slice(self, enrolled_pool): + enrolled_pool.get_or_allocate("testnode_0", 0, MiB) + enrolled_pool.get_or_allocate("testnode_0", 1, MiB) + enrolled_pool.release_owner("testnode_0") + reused_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + assert reused_offset == 0 + + def test_different_sizes_pack_sequentially(self, enrolled_pool): + first_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + second_offset = enrolled_pool.get_or_allocate("testnode_0", 0, 2 * MiB) + assert (first_offset, second_offset) == (0, MiB) + + def test_parent_change_rejected(self, tmp_path, fake_host, enrolled_pool): + other = fake_host.add_block_device("nvme1n1p1") + with pytest.raises(BlockPoolError, match="parent changed"): + BlockPool(tmp_path / "block_pool.json", other, allow_enroll=True) + + def test_enroll_requires_allow_enroll(self, disallow_enroll_pool): + with pytest.raises(BlockPoolError, match="allow_enroll"): + disallow_enroll_pool.ensure_ready() + + @pytest.mark.parametrize( + "cfg_allow_enroll,expected", + [ + ({}, False), # defaults to False when omitted + ({"allow_enroll": False}, False), + ({"allow_enroll": True}, True), + ], + ) + def test_from_config_allow_enroll( + self, cfg_allow_enroll, expected, tmp_path, pool_parent + ): + pool = BlockPool.from_config( + { + "block_pool": { + "parent": pool_parent, + "state_dir": str(tmp_path), + **cfg_allow_enroll, + } + } + ) + assert pool is not None + assert pool.allow_enroll is expected + + def test_state_dir_defaults_to_data_dir(self, tmp_path, pool_parent): + pool = BlockPool.from_config( + {"data_dir": str(tmp_path), "block_pool": {"parent": pool_parent}} + ) + assert pool is not None + assert pool.state_path == tmp_path / "block_pool.json" + + def test_shared_pool_is_built_once(self, tmp_path, pool_parent): + config["block_pool"] = {"parent": pool_parent, "state_dir": str(tmp_path)} + pool = shared_pool() + assert pool is not None and pool.parent == pool_parent + assert shared_pool() is pool + + def test_enroll_rejects_non_empty_parent(self, fake_host, block_pool): + write(fake_host, block_pool.parent, 0, b"DATA") + with pytest.raises(BlockPoolError, match="not empty"): + block_pool.ensure_ready() + + def test_enroll_writes_marker_and_state(self, block_pool): + block_pool.ensure_ready() + assert block_pool.enrolled + assert block_pool._read_tail_marker() == { + "pool_id": block_pool.pool_id, + "parent": block_pool.parent, + } + + def test_new_slice_rejects_existing_data(self, fake_host, enrolled_pool): + write(fake_host, enrolled_pool.parent, 0, b"NOTZEROS") + with pytest.raises(BlockPoolError, match="not empty"): + enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + + def test_enroll_rejects_whole_disk_with_partitions(self, tmp_path, fake_host): + parent = fake_host.add_block_device("nvme0n1") + fake_host.add_block_device("nvme0n1p1", MiB, disk="nvme0n1") + pool = BlockPool(tmp_path / "block_pool.json", parent, allow_enroll=True) + with pytest.raises(BlockPoolError, match="existing partitions"): + pool.ensure_ready() + + def test_reclaims_marker_after_state_loss(self, disallow_enroll_pool): + payload = {"pool_id": "saved-pool-id", "parent": disallow_enroll_pool.parent} + disallow_enroll_pool._write_tail_marker(payload) + disallow_enroll_pool.ensure_ready() + assert disallow_enroll_pool.enrolled + assert disallow_enroll_pool.pool_id == "saved-pool-id" + + def test_reused_slice_must_be_empty(self, fake_host, enrolled_pool): + enrolled_pool.get_or_allocate("testnode_0", 0, MiB) + enrolled_pool.release_owner("testnode_0") + write(fake_host, enrolled_pool.parent, 0, b"STALE") + with pytest.raises(BlockPoolError, match="not empty"): + enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + + def test_released_region_is_zeroed_and_reusable(self, fake_host, enrolled_pool): + parent = enrolled_pool.parent + assert enrolled_pool.get_or_allocate("testnode_0", 0, MiB) == 0 + write(fake_host, parent, 0, b"BLUESTORE LABEL") # the owner used it + fake_host.commands.clear() + assert enrolled_pool.release_owner("testnode_0") == [] + writes = [c for c in fake_host.commands if any("of=" in a for a in c)] + assert writes == [ + ["dd", f"of={parent}", "bs=4096", "seek=0", "conv=notrunc,fsync"] + ] + assert bytes(fake_host.files[parent][:4096]) == bytes(4096) + assert enrolled_pool.get_or_allocate("ceph_node", 0, MiB) == 0 + + def test_release_zeroes_bluestore_label_copies(self, fake_host, enrolled_pool): + parent = enrolled_pool.parent + # Label copies at 256 KiB and 512 KiB fall inside a 1 MiB region; the + # one at 1 MiB is the next region's start and must not be touched. + labels = (0, 256 * 1024, 512 * 1024, MiB) + with patch("ceph_devstack.block_pool._BLUESTORE_LABEL_OFFSETS", labels): + assert enrolled_pool.get_or_allocate("testnode_0", 0, MiB) == 0 + assert enrolled_pool.get_or_allocate("testnode_1", 0, MiB) == MiB + for label in labels: + write(fake_host, parent, label, b"bluestore block device") + fake_host.commands.clear() + assert enrolled_pool.release_owner("testnode_0") == [] + seeks = [ + a + for c in fake_host.commands + if any("of=" in a for a in c) + for a in c + if a.startswith("seek=") + ] + assert seeks == ["seek=0", "seek=64", "seek=128"] + for label in labels[:3]: + assert bytes(fake_host.files[parent][label : label + 4096]) == bytes( + 4096 + ) + assert fake_host.files[parent][MiB : MiB + 9] == b"bluestore" + + def test_new_slice_rejects_bluestore_label_copy(self, fake_host, enrolled_pool): + with patch( + "ceph_devstack.block_pool._BLUESTORE_LABEL_OFFSETS", (0, 512 * 1024) + ): + write(fake_host, enrolled_pool.parent, 512 * 1024, b"bluestore") + with pytest.raises(BlockPoolError, match=f"at offset {512 * 1024}"): + enrolled_pool.get_or_allocate("ceph_node", 0, MiB) + + def test_region_that_cannot_be_zeroed_stays_allocated( + self, fake_host, enrolled_pool + ): + enrolled_pool.get_or_allocate("testnode_0", 0, MiB) + with patch.object( + enrolled_pool, "_write_parent", side_effect=BlockPoolError("EIO") + ): + assert enrolled_pool.release_owner("testnode_0") == ["testnode_0:0"] + assert list(enrolled_pool.allocations) == ["testnode_0:0"] + assert enrolled_pool.free_regions == [] + + @pytest.mark.parametrize("marker", ["missing", "foreign", "unenrolled"]) + def test_release_writes_nothing_unless_parent_is_ours( + self, fake_host, enrolled_pool, marker + ): + enrolled_pool.get_or_allocate("testnode_0", 0, MiB) + enrolled_pool.get_or_allocate("testnode_0", 1, MiB) + if marker == "missing": + write(fake_host, enrolled_pool.parent, TAIL, bytes(4096)) + elif marker == "foreign": + enrolled_pool._write_tail_marker( + {"parent": enrolled_pool.parent, "pool_id": "other-pool"} + ) + else: + state = json.loads(enrolled_pool.state_path.read_text()) + state["enrolled"] = False + enrolled_pool.state_path.write_text(json.dumps(state)) + write(fake_host, enrolled_pool.parent, 0, b"FOREIGN") + fake_host.commands.clear() + kept = enrolled_pool.release_owner("testnode_0") + assert kept == ["testnode_0:0", "testnode_0:1"] + assert not [c for c in fake_host.commands if any("of=" in a for a in c)] + assert bytes(fake_host.files[enrolled_pool.parent][:7]) == b"FOREIGN" + assert sorted(enrolled_pool.allocations) == kept + + +class TestBlockPoolEnrollmentGuards: + def test_enroll_rejects_mounted_parent(self, fake_host, block_pool): + fake_host.responses["findmnt"] = (0, b"/mnt\n") + with pytest.raises(BlockPoolError, match="mounted"): + block_pool.ensure_ready() + assert [ + "findmnt", + "-n", + "-o", + "TARGET", + "--source", + block_pool.parent, + ] in fake_host.commands + + def test_enroll_rejects_blkid_signature(self, fake_host, block_pool): + fake_host.responses["blkid"] = (0, b"DEVNAME=x\nTYPE=xfs\n") + with pytest.raises(BlockPoolError, match="filesystem or partition"): + block_pool.ensure_ready() + + def test_verify_rejects_missing_marker(self, fake_host, enrolled_pool): + write(fake_host, enrolled_pool.parent, TAIL, bytes(4096)) + with pytest.raises(BlockPoolError, match="missing the devstack"): + enrolled_pool.ensure_ready() + + def test_verify_rejects_foreign_pool_id(self, enrolled_pool): + enrolled_pool._write_tail_marker( + {"parent": enrolled_pool.parent, "pool_id": "other-pool"} + ) + with pytest.raises(BlockPoolError, match="belongs to another pool"): + enrolled_pool.ensure_ready() + + def test_region_exceeds_parent_size(self, enrolled_pool): + with pytest.raises(BlockPoolError, match="exceeds"): + enrolled_pool.get_or_allocate("ceph_node", 0, 32 * MiB) + + +class TestHasPartitions: + def test_whole_disk_with_partitions(self, fake_host): + fake_host.add_block_device("nvme0n1") + fake_host.add_block_device("nvme0n1p1", disk="nvme0n1") + assert BlockPool._has_partitions("nvme0n1") is True + + @pytest.mark.parametrize( + "disk,other_disk,other_partition", + [("nvme0n1", "nvme0n11", "nvme0n11p1"), ("sda", "sdaa", "sdaa1")], + ) + def test_partitions_of_similarly_named_disks_do_not_count( + self, fake_host, disk, other_disk, other_partition + ): + fake_host.add_block_device(disk) + fake_host.add_block_device(other_disk) + fake_host.add_block_device(other_partition, disk=other_disk) + assert BlockPool._has_partitions(disk) is False + + def test_partition_has_no_partitions(self): + assert BlockPool._has_partitions("nvme0n1p1") is False + + +class TestTailMarker: + def test_roundtrip(self, block_pool): + payload = {"pool_id": "abc", "parent": block_pool.parent} + block_pool._write_tail_marker(payload) + assert block_pool._read_tail_marker() == payload + + def test_corrupt_marker_returns_none(self, fake_host, block_pool): + write( + fake_host, block_pool.parent, TAIL, b"CEPH-DEVSTACK-BLOCK-POOL-TAIL-v1\x00x" + ) + assert block_pool._read_tail_marker() is None diff --git a/tests/test_host.py b/tests/test_host.py index 6588576e..246d09eb 100644 --- a/tests/test_host.py +++ b/tests/test_host.py @@ -1,3 +1,7 @@ +import os +import subprocess +from unittest.mock import MagicMock, patch + import pytest from ceph_devstack.host import LocalHost, RemoteHost @@ -12,11 +16,64 @@ def test_path_exists_expands_tilde(self, host): assert host.path_exists("~") is True assert host.path_exists("~/nonexistent_dir_12345") is False + def test_file_helpers_run_commands_on_host(self, host, tmp_path): + (tmp_path / "a").write_text("hello\n") + (tmp_path / "link").symlink_to(tmp_path / "a") + assert host.read_text(tmp_path / "a") == "hello\n" + assert host.read_text(tmp_path / "missing") is None + assert host.list_dir(tmp_path) == ["a", "link"] + assert host.list_dir(tmp_path / "missing") == [] + assert host.realpath(tmp_path / "link") == os.path.realpath(tmp_path / "a") + + def test_byte_io_at_unaligned_offsets(self, host, tmp_path): + disk = tmp_path / "disk" + disk.write_bytes(bytes(8192)) + host.write_bytes(disk, 4096 + 256, b"x" * 512) + data = disk.read_bytes() + assert len(data) == 8192 + assert data[4352:4864] == b"x" * 512 and data.count(b"x") == 512 + assert host.read_bytes(disk, 4096, 1024) == bytes(256) + b"x" * 512 + bytes(256) + + @pytest.mark.parametrize("stdout,mounted", [(b"/mnt\n", True), (b"", False)]) + def test_device_mounted_asks_findmnt(self, host, stdout, mounted): + done = subprocess.CompletedProcess([], 0 if stdout else 1, stdout, b"") + with patch.object(host, "output", return_value=done) as output: + assert host.device_mounted("/dev/loop3") is mounted + output.assert_called_once_with( + ["findmnt", "-n", "-o", "TARGET", "--source", "/dev/loop3"] + ) + + def test_user_ids_returns_os_ids(self, host): + uid, gid = host.user_ids() + assert uid == os.getuid() + assert gid == os.getgid() + assert isinstance(uid, int) + assert isinstance(gid, int) + SSH = ["podman", "machine", "ssh", "--"] class TestRemoteHost: + def test_write_bytes_pipes_data_through_machine_ssh(self): + remote = RemoteHost() + done = MagicMock(returncode=0) + with patch("ceph_devstack.exec.subprocess.run", return_value=done) as run: + remote.write_bytes("/dev/sda1", 8192, b"m" * 4096) + args, kwargs = run.call_args + assert args[0] == [ + "podman", + "machine", + "ssh", + "--", + "dd", + "of=/dev/sda1", + "bs=4096", + "seek=2", + "conv=notrunc,fsync", + ] + assert kwargs["input"] == b"m" * 4096 + @pytest.mark.parametrize( "args,expected", [ @@ -30,3 +87,27 @@ class TestRemoteHost: ) def test_cmd_runs_non_podman_commands_in_machine(self, args, expected): assert RemoteHost().cmd(args).args == expected + + def test_user_ids_parses_and_caches(self): + remote = RemoteHost() + outputs = {"-u": b"501\n", "-g": b"1000\n"} + + def output(args): + return subprocess.CompletedProcess(args, 0, outputs[args[1]], b"") + + with patch.object(remote, "output", side_effect=output) as mock_output: + assert remote.user_ids() == (501, 1000) + assert remote.user_ids() == (501, 1000) + assert [c.args[0] for c in mock_output.call_args_list] == [ + ["id", "-u"], + ["id", "-g"], + ] + + def test_user_ids_raises_on_invalid_output(self): + remote = RemoteHost() + bad = subprocess.CompletedProcess([], 0, b"not-a-number\n", b"") + with ( + patch.object(remote, "output", return_value=bad), + pytest.raises(RuntimeError, match="'id -u'"), + ): + remote.user_ids() From 4617e1b12040f326000e42f928730573a839c11e Mon Sep 17 00:00:00 2001 From: Zack Cerza Date: Wed, 21 Oct 2026 14:30:57 -0600 Subject: [PATCH 2/3] TestNode: Use LoopDeviceMixin TestNode derived loop device numbers from its index, so they could collide with loop devices already in use on the host. Use LoopDeviceMixin instead, which also lets testnodes use the block pool. The stack allocates loop numbers for every container before creating them in parallel, so concurrent creates can't pick the same ones. Signed-off-by: Zack Cerza --- ceph_devstack/config.toml | 1 + ceph_devstack/resources/ceph/__init__.py | 7 ++ ceph_devstack/resources/ceph/containers.py | 75 +---------------- .../resources/ceph/test_cephdevstack_core.py | 17 +++- tests/resources/ceph/test_testnode.py | 80 ++++++++++--------- tests/test_config.py | 1 + 6 files changed, 72 insertions(+), 109 deletions(-) diff --git a/ceph_devstack/config.toml b/ceph_devstack/config.toml index 147a729e..38c2d892 100644 --- a/ceph_devstack/config.toml +++ b/ceph_devstack/config.toml @@ -17,6 +17,7 @@ image = "quay.io/ceph-infra/pulpito:main" [containers.testnode] count = 3 +loop_device_count = 1 loop_device_size = "5G" image = "quay.io/ceph-infra/teuthology-testnode:main" diff --git a/ceph_devstack/resources/ceph/__init__.py b/ceph_devstack/resources/ceph/__init__.py index 5864c1b6..ec3b1cac 100644 --- a/ceph_devstack/resources/ceph/__init__.py +++ b/ceph_devstack/resources/ceph/__init__.py @@ -18,6 +18,7 @@ Teuthology, Archive, ) +from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin from ceph_devstack.resources.ceph.requirements import ( HasSudo, LoopControlDeviceExists, @@ -158,6 +159,12 @@ async def create(self): logger.info("Creating containers...") await CephDevStackNetwork().create() await SSHKeyPair().create() + # Pre-allocate loop device numbers sequentially so concurrent creates + # don't race on the same device numbers. + for spec in self.service_specs.values(): + for obj in spec["objects"]: + if isinstance(obj, LoopDeviceMixin): + obj.allocate_devices() containers = [] for spec in self.service_specs.values(): for object in spec["objects"]: diff --git a/ceph_devstack/resources/ceph/containers.py b/ceph_devstack/resources/ceph/containers.py index 1d3807c7..6a5f3ac3 100644 --- a/ceph_devstack/resources/ceph/containers.py +++ b/ceph_devstack/resources/ceph/containers.py @@ -6,6 +6,7 @@ from ceph_devstack import config, DEFAULT_CONFIG_PATH, logger from ceph_devstack.host import host +from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin from ceph_devstack.resources.container import Container @@ -154,7 +155,7 @@ class Pulpito(Container): } -class TestNode(Container): +class TestNode(LoopDeviceMixin, Container): _image_name = "teuthology-testnode" capabilities = [ "SYS_ADMIN", @@ -178,13 +179,7 @@ class TestNode(Container): def __init__(self, name: str = ""): super().__init__(name=name) - self.index = 0 - if "_" in self.name: - self.index = int(self.name.split("_")[-1]) - self.loop_device_count = config["containers"]["testnode"].get( - "loop_device_count", 1 - ) - self.devices = [self.device_name(i) for i in range(self.loop_device_count)] + self._init_loop_devices() @property def loop_img_dir(self): @@ -260,70 +255,6 @@ async def remove(self): await super().remove() await self.remove_loop_devices() - async def create_loop_devices(self): - for device in self.devices: - await self.create_loop_device(device) - - async def remove_loop_devices(self): - for device in self.devices: - await self.remove_loop_device(device) - - async def create_loop_device(self, device: str): - size = config["containers"]["testnode"]["loop_device_size"] - os.makedirs(self.loop_img_dir, exist_ok=True) - proc = await self.cmd(["lsmod", "|", "grep", "loop"]) - if proc and await proc.wait() != 0: - await self.cmd(["sudo", "modprobe", "loop"]) - loop_img_name = os.path.join(self.loop_img_dir, self.device_image(device)) - await self.remove_loop_device(device) - device_pos = device.removeprefix("/dev/loop") - await self.cmd( - [ - "sudo", - "mknod", - "-m700", - device, - "b", - "7", - device_pos, - ], - check=True, - ) - await self.cmd( - ["sudo", "chown", f"{os.getuid()}:{os.getgid()}", device], - check=True, - ) - await self.cmd( - [ - "sudo", - "dd", - "if=/dev/null", - f"of={loop_img_name}", - "bs=1", - "count=0", - f"seek={size}", - ], - check=True, - ) - await self.cmd(["sudo", "losetup", device, loop_img_name], check=True) - await self.cmd(["chcon", "-t", "fixed_disk_device_t", device]) - - async def remove_loop_device(self, device: str): - loop_img_name = os.path.join(self.loop_img_dir, self.device_image(device)) - if os.path.ismount(device): - await self.cmd(["umount", device], check=True) - if host.path_exists(device): - await self.cmd(["sudo", "losetup", "-d", device]) - await self.cmd(["sudo", "rm", "-f", device], check=True) - if host.path_exists(loop_img_name): - os.remove(loop_img_name) - - def device_name(self, index: int): - return f"/dev/loop{self.loop_device_count * self.index + index}" - - def device_image(self, device: str): - return f"{self.name}-{device.removeprefix('/dev/loop')}" - def _get_available_dmi_mounts(self) -> List[str]: """Return volume mount args for DMI files that exist on the host. diff --git a/tests/resources/ceph/test_cephdevstack_core.py b/tests/resources/ceph/test_cephdevstack_core.py index fa22590e..ddf2d889 100644 --- a/tests/resources/ceph/test_cephdevstack_core.py +++ b/tests/resources/ceph/test_cephdevstack_core.py @@ -61,6 +61,17 @@ def test_service_specs_named_objects_when_count_greater_than_one(self): assert testnode_objects[1].name == "testnode_1" assert testnode_objects[2].name == "testnode_2" + def test_multiple_testnodes_get_distinct_loop_devices(self, tmp_path): + config["data_dir"] = str(tmp_path) + devstack = CephDevStack() + testnode_objects = devstack.service_specs["testnode"]["objects"] + all_devices = [] + for node in testnode_objects: + all_devices.extend(node.devices) + assert len(all_devices) == len(set(all_devices)), ( + f"Loop device collision: {all_devices}" + ) + def test_service_specs_sets_postgres_paddles_url(self): devstack = CephDevStack() paddles_obj = devstack.service_specs["paddles"]["objects"][0] @@ -439,7 +450,11 @@ def test_init_without_postgres(self): "paddles": {"image": "paddles:latest", "count": 1}, "beanstalk": {"image": "beanstalk:latest", "count": 1}, "pulpito": {"image": "pulpito:latest", "count": 1}, - "testnode": {"image": "testnode:latest", "count": 3}, + "testnode": { + "image": "testnode:latest", + "count": 3, + "loop_device_count": 1, + }, "teuthology": {"image": "teuthology:latest", "count": 1}, "archive": {"image": "archive:latest", "count": 1}, } diff --git a/tests/resources/ceph/test_testnode.py b/tests/resources/ceph/test_testnode.py index 6971afc5..53bfe2c7 100644 --- a/tests/resources/ceph/test_testnode.py +++ b/tests/resources/ceph/test_testnode.py @@ -1,54 +1,60 @@ from pathlib import Path from unittest.mock import patch -import pytest - -from ceph_devstack.resources.ceph import TestNode as _TestNode from ceph_devstack import config +from ceph_devstack.resources.ceph import TestNode as _TestNode class TestTestnode: - @pytest.fixture(scope="class") - @classmethod - def cls(self) -> type[_TestNode]: - return _TestNode - - def test_testnode_loop_device_count_default_to_one(self, cls): - testnode = cls("testnode_1") + def test_testnode_loop_device_count_default_to_one(self): + testnode = _TestNode("testnode_1") assert testnode.loop_device_count == 1 - def test_testnode_create_cmd_includes_related_devices(self, cls): + def test_testnode_create_cmd_includes_related_devices(self, tmp_path): config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml")) - testnode = cls("testnode_1") + config["data_dir"] = str(tmp_path) + testnode = _TestNode("testnode_1") create_cmd = testnode.create_cmd - assert "--device=/dev/loop4" in create_cmd - assert "--device=/dev/loop5" in create_cmd - assert "--device=/dev/loop6" in create_cmd - assert "--device=/dev/loop7" in create_cmd + assert "--device=/dev/loop0" in create_cmd + assert "--device=/dev/loop1" in create_cmd + assert "--device=/dev/loop2" in create_cmd + assert "--device=/dev/loop3" in create_cmd - def test_testnode_create_cmd_mounts_run_udev_read_only(self, cls): - create_cmd = cls("testnode_1").create_cmd + def test_testnode_create_cmd_mounts_run_udev_read_only(self, tmp_path): + config["data_dir"] = str(tmp_path) + create_cmd = _TestNode("testnode_1").create_cmd index = create_cmd.index("/run/udev:/run/udev:ro") assert create_cmd[index - 1] == "-v" - def test_testnode_create_cmd_unmasks_proc_for_nested_containers(self, cls): - create_cmd = cls("testnode_1").create_cmd + def test_testnode_create_cmd_unmasks_proc_for_nested_containers(self, tmp_path): + config["data_dir"] = str(tmp_path) + create_cmd = _TestNode("testnode_1").create_cmd index = create_cmd.index("unmask=/proc/*") assert create_cmd[index - 1] == "--security-opt" - def test_testnode_devices_is_based_on_loop_device_count_config(self, cls): + def test_testnode_devices_is_based_on_loop_device_count_config(self, tmp_path): config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml")) - testnode = cls("testnode_1") + config["data_dir"] = str(tmp_path) + testnode = _TestNode("testnode_1") assert testnode.loop_device_count == 4 assert testnode.devices == [ - "/dev/loop4", - "/dev/loop5", - "/dev/loop6", - "/dev/loop7", + "/dev/loop0", + "/dev/loop1", + "/dev/loop2", + "/dev/loop3", ] - def test_get_available_dmi_mounts_with_all_files_present(self, cls): - testnode = cls("testnode_1") + def test_testnode_reuses_existing_backing_files(self, tmp_path, fake_host): + config["data_dir"] = str(tmp_path) + image_dir = tmp_path / "disk_images" + fake_host.add_file(image_dir / "testnode_1-4") + fake_host.add_file(image_dir / "testnode_1-1") + config["containers"]["testnode"]["loop_device_count"] = 3 + testnode = _TestNode("testnode_1") + assert testnode.devices == ["/dev/loop1", "/dev/loop4", "/dev/loop0"] + + def test_get_available_dmi_mounts_with_all_files_present(self): + testnode = _TestNode("testnode_1") with patch( "ceph_devstack.resources.ceph.containers.host.path_exists" ) as mock_exists: @@ -63,8 +69,8 @@ def test_get_available_dmi_mounts_with_all_files_present(self, cls): "/dev/null:/sys/class/dmi/id/product_serial", ] - def test_get_available_dmi_mounts_with_no_files_present(self, cls): - testnode = cls("testnode_1") + def test_get_available_dmi_mounts_with_no_files_present(self): + testnode = _TestNode("testnode_1") with patch( "ceph_devstack.resources.ceph.containers.host.path_exists" ) as mock_exists: @@ -72,8 +78,8 @@ def test_get_available_dmi_mounts_with_no_files_present(self, cls): mounts = testnode._get_available_dmi_mounts() assert mounts == [] - def test_get_available_dmi_mounts_with_partial_files_present(self, cls): - testnode = cls("testnode_1") + def test_get_available_dmi_mounts_with_partial_files_present(self): + testnode = _TestNode("testnode_1") with patch( "ceph_devstack.resources.ceph.containers.host.path_exists" ) as mock_exists: @@ -85,9 +91,10 @@ def test_get_available_dmi_mounts_with_partial_files_present(self, cls): "/dev/null:/sys/class/dmi/id/board_serial", ] - def test_create_cmd_includes_dmi_mounts_when_available(self, cls): + def test_create_cmd_includes_dmi_mounts_when_available(self, tmp_path): config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml")) - testnode = cls("testnode_1") + config["data_dir"] = str(tmp_path) + testnode = _TestNode("testnode_1") with patch( "ceph_devstack.resources.ceph.containers.host.path_exists" ) as mock_exists: @@ -97,9 +104,10 @@ def test_create_cmd_includes_dmi_mounts_when_available(self, cls): assert "/dev/null:/sys/class/dmi/id/chassis_serial" in " ".join(create_cmd) assert "/dev/null:/sys/class/dmi/id/product_serial" in " ".join(create_cmd) - def test_create_cmd_excludes_dmi_mounts_when_unavailable(self, cls): + def test_create_cmd_excludes_dmi_mounts_when_unavailable(self, tmp_path): config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml")) - testnode = cls("testnode_1") + config["data_dir"] = str(tmp_path) + testnode = _TestNode("testnode_1") with patch( "ceph_devstack.resources.ceph.containers.host.path_exists" ) as mock_exists: diff --git a/tests/test_config.py b/tests/test_config.py index 5d11c1dc..cd4ed7d9 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -114,6 +114,7 @@ def test_config_defaults(self): "pulpito": {"image": "quay.io/ceph-infra/pulpito:main"}, "testnode": { "count": 3, + "loop_device_count": 1, "loop_device_size": "5G", "image": "quay.io/ceph-infra/teuthology-testnode:main", }, From f18bc4be5e5e649f099d62613a5b234b28c19ec2 Mon Sep 17 00:00:00 2001 From: Zack Cerza Date: Wed, 21 Oct 2026 14:31:20 -0600 Subject: [PATCH 3/3] Add block-pool config, status command and doctor checks Ship a commented [block_pool] section, add `ceph-devstack block-pool status` to show allocations, and have doctor check that the user can read and write the pool's parent device. Document the pool in the README. On Fedora CoreOS (e.g. a podman machine) the disk group is only defined in /usr/lib/group, where usermod can't add members, so doctor --fix copies it to /etc/group first. Signed-off-by: Zack Cerza --- README.md | 51 +++++ ceph_devstack/__init__.py | 5 + ceph_devstack/cli.py | 45 +++- ceph_devstack/config.toml | 14 ++ ceph_devstack/resources/ceph/__init__.py | 4 + ceph_devstack/resources/ceph/requirements.py | 76 ++++++- .../resources/ceph/test_requirements_ceph.py | 210 ++++++++++-------- tests/test_cli.py | 49 ++++ tests/test_config.py | 1 + 9 files changed, 361 insertions(+), 94 deletions(-) create mode 100644 tests/test_cli.py diff --git a/README.md b/README.md index 488d973c..77b75619 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,57 @@ python3 -m pip install git+https://github.com/zmc/ceph-devstack.git `ceph-devstack config dump` will output the current configuration. +### Shared block pool (optional) + +To back loop devices with slices of a real NVMe device instead of sparse files, +configure `[block_pool]` in your config. The pool is shared across all +`testnode_*` containers. Each loop device uses that container's +`loop_device_size` for its backing region on the pool parent. + +```toml +[block_pool] +parent = "/dev/nvme0n1p1" # or a dedicated whole disk +allow_enroll = true # first run only; unset afterward + +[containers.testnode] +loop_device_size = "5G" +``` + +Requirements: + +* Your user must be in the `disk` group and able to read/write the parent + device directly (enrollment writes a tail marker to the raw device). +* On a Fedora CoreOS podman machine (macOS), the `disk` group is defined in + `/usr/lib/group`, so `usermod -aG disk core` silently does nothing until the + group line is copied into `/etc/group`. `ceph-devstack doctor --fix` does + both; by hand, inside `podman machine ssh`: + + ```bash + grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group | sudo tee -a /etc/group + sudo usermod -aG disk core + ``` + + New `podman machine ssh` sessions then have the group. +* The parent must be empty, or already enrolled by ceph-devstack (tail marker + at the end of the device). + +`ceph-devstack block-pool status` shows current allocations. + +All pool device access (sysfs, `findmnt`, `blkid`, and the marker and probe +reads/writes via `dd`) runs on the container host, so a pool inside a podman +machine (macOS, `RemoteHost`) should work, but that setup is unverified. The +pool state file and its lock always stay on the local machine. + +Each container picks host loop devices at create time: it needs +`loop_device_count` devices of `loop_device_size`, reuses any existing backing +files for that container name, then takes the lowest free loop numbers after +scanning what is already attached or claimed in the `disk_images` directory. + +If `block_pool.json` is lost but the on-disk marker remains, start again to +reclaim the marker; the device is not reformatted. The old allocations are +forgotten, though, so regions that still hold data are refused as not empty +until they are wiped. + As an example, the following configuration will use a local image for paddles with the tag `TEST`; it will also create ten testnode containers; and will build its teuthology container from the git repo at `~/src/teuthology`: ``` containers: diff --git a/ceph_devstack/__init__.py b/ceph_devstack/__init__.py index 9eae8d52..71612111 100644 --- a/ceph_devstack/__init__.py +++ b/ceph_devstack/__init__.py @@ -62,6 +62,11 @@ def parse_args(args: List[str]) -> argparse.Namespace: default=False, help="Apply suggested fixes for issues found", ) + parser_block_pool = subparsers.add_parser( + "block-pool", help="Inspect the shared block device pool" + ) + block_pool_subparsers = parser_block_pool.add_subparsers(dest="block_pool_op") + block_pool_subparsers.add_parser("status", help="Show pool allocations and state") parser_pull = subparsers.add_parser("pull", help="Pull container images") parser_pull.add_argument( "image", diff --git a/ceph_devstack/cli.py b/ceph_devstack/cli.py index 8658bcbc..760b5490 100644 --- a/ceph_devstack/cli.py +++ b/ceph_devstack/cli.py @@ -5,6 +5,7 @@ from pathlib import Path from ceph_devstack import config, logger, parse_args, VERBOSE +from ceph_devstack.block_pool import BlockPoolError, format_size, shared_pool from ceph_devstack.requirements import check_requirements from ceph_devstack.resources.ceph import CephDevStack @@ -32,16 +33,52 @@ } +def _configure_console_logging(verbose: bool) -> None: + if not verbose: + return + for handler in logging.getLogger("root").handlers: + if not isinstance(handler, logging.FileHandler): + handler.setLevel(VERBOSE) + + +def block_pool_status() -> int: + try: + pool = shared_pool() + except BlockPoolError as e: + logger.error(f"block pool: {e}") + return 1 + if pool is None: + logger.info("block pool: disabled (no parent configured)") + return 0 + logger.info(f"block pool parent: {pool.parent}") + logger.info(f"enrolled: {pool.enrolled}") + logger.info(f"pool id: {pool.pool_id}") + logger.info(f"state file: {pool.state_path}") + logger.info("allocations:" if pool.allocations else "allocations: (none)") + for key, region in sorted(pool.allocations.items()): + logger.info( + f" {key}: offset {region['offset']}, " + f"size {format_size(region['size'])}, last owner {region['owner']}" + ) + logger.info("free regions:" if pool.free_regions else "free regions: (none)") + for region in pool.free_regions: + logger.info(f" offset {region['offset']}, size {format_size(region['size'])}") + logger.info(f"next offset: {pool.next_offset}") + return 0 + + def main() -> int: args = parse_args(sys.argv[1:]) config.load(args.config_file) - if args.verbose: - for handler in logging.getLogger("root").handlers: - if not isinstance(handler, logging.FileHandler): - handler.setLevel(VERBOSE) + _configure_console_logging(args.verbose) if args.command == "config": CONFIG_HANDLERS[args.config_op](config, args) return 0 + if args.command == "block-pool": + if args.block_pool_op != "status": + logger.error("Usage: ceph-devstack block-pool status") + return 2 + return block_pool_status() config["args"] = vars(args) data_path = Path(config["data_dir"]).expanduser() data_path.mkdir(parents=True, exist_ok=True) diff --git a/ceph_devstack/config.toml b/ceph_devstack/config.toml index 38c2d892..41f24b1d 100644 --- a/ceph_devstack/config.toml +++ b/ceph_devstack/config.toml @@ -1,5 +1,19 @@ data_dir = "~/.local/share/ceph-devstack" +[block_pool] +# Dedicated empty block device or partition. Whole disks are allowed when +# empty; disks with existing partitions must use a partition path instead. +# Enrollment reads/writes the parent directly: your user must be able to +# open the device for read/write (typically membership in the disk group). +# First run requires allow_enroll = true; the tool writes a tail marker and +# refuses non-empty or foreign devices. If block_pool.json is lost but the +# marker remains, start again to reclaim the marker; the old allocations are +# forgotten, so regions that still hold data are refused until wiped. +# parent = "/dev/nvme0n1" +# parent = "/dev/nvme0n1p1" +# allow_enroll = true +# state_dir = "/path" # where block_pool.json lives; defaults to data_dir + [containers.archive] image = "python:alpine" diff --git a/ceph_devstack/resources/ceph/__init__.py b/ceph_devstack/resources/ceph/__init__.py index ec3b1cac..bbe10451 100644 --- a/ceph_devstack/resources/ceph/__init__.py +++ b/ceph_devstack/resources/ceph/__init__.py @@ -23,6 +23,8 @@ HasSudo, LoopControlDeviceExists, LoopControlDeviceWriteable, + BlockPoolDiskGroup, + BlockPoolParentAccessible, SELinuxModule, ) from ceph_devstack.resources.ceph.utils import get_runs, get_jobs @@ -129,6 +131,8 @@ async def check_requirements(self): result = has_sudo = await HasSudo().evaluate() result = result and await LoopControlDeviceExists().evaluate() result = result and await LoopControlDeviceWriteable().evaluate() + result = result and await BlockPoolDiskGroup().evaluate() + result = result and await BlockPoolParentAccessible().evaluate() # Check for SELinux being enabled and Enforcing; then check for the # presence of our module. If necessary, inform the user and instruct diff --git a/ceph_devstack/resources/ceph/requirements.py b/ceph_devstack/resources/ceph/requirements.py index 7a19d3e9..aee63c49 100644 --- a/ceph_devstack/resources/ceph/requirements.py +++ b/ceph_devstack/resources/ceph/requirements.py @@ -1,12 +1,19 @@ import hashlib import re +import shlex from subprocess import CalledProcessError from ceph_devstack import logger, PROJECT_ROOT +from ceph_devstack.block_pool import BlockPool, BlockPoolError, shared_pool +from ceph_devstack.host import Host from ceph_devstack.requirements import Requirement, FixableRequirement +def _host_user(host: Host) -> str: + return host.output(["whoami"]).stdout.decode().strip() + + class HasSudo(Requirement): check_cmd = ["sudo", "true"] suggest_msg = "sudo access is required" @@ -31,7 +38,7 @@ async def check(self): .communicate()[0] .decode() ) - user = self.host.run(["whoami"]).communicate()[0].strip().decode() + user = _host_user(self.host) if self.host.type == "local": self.fix_cmd = ["sudo", "usermod", "-a", "-G", group, user] else: @@ -47,6 +54,73 @@ async def suggest(self): ) +class BlockPoolRequirement(Requirement): + """Passes without a configured pool; fails (logged) if its config is invalid. + + An invalid config fails in evaluate(), before any suggest or --fix step, + since there is nothing to fix on the host for it. + """ + + async def evaluate(self) -> bool: + try: + pool = shared_pool() + except BlockPoolError as e: + logger.error(f"block pool: {e}") + return False + return pool is None or await super().evaluate() + + async def check(self) -> bool: + pool = shared_pool() + return pool is None or await self.check_pool(pool) + + async def check_pool(self, pool: BlockPool) -> bool: + raise NotImplementedError + + +# Fedora CoreOS (e.g. a podman machine) defines the disk group in +# /usr/lib/group, where usermod silently fails to add members; copy the group +# into /etc/group first. Elsewhere it is already there and this does nothing. +DISK_GROUP_TO_ETC = ( + "grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group >> /etc/group" +) + + +class BlockPoolDiskGroup(BlockPoolRequirement, FixableRequirement): + check_cmd = ["bash", "-c", "id -nG | grep -qw disk"] + suggest_msg = ( + "block pool parent requires membership in the disk group " + "(log out and back in after joining)" + ) + + async def check_pool(self, pool: BlockPool) -> bool: + if await Requirement.check(self): + return True + user = shlex.quote(_host_user(self.host)) + self.fix_cmd = [ + "sudo", + "bash", + "-c", + f"{DISK_GROUP_TO_ETC}; usermod -a -G disk {user}", + ] + return False + + +class BlockPoolParentAccessible(BlockPoolRequirement): + suggest_msg = "block pool parent must be readable and writable by this user" + + async def check_pool(self, pool: BlockPool) -> bool: + proc = await self.host.arun( + ["test", "-r", pool.parent, "-a", "-w", pool.parent] + ) + if await proc.wait() == 0: + return True + logger.error( + f"{self.suggest_msg} ({pool.parent}). " + "Join the disk group and re-login, or fix device permissions." + ) + return False + + class SELinuxModule(FixableRequirement): MODULE_NAME = "ceph_devstack" SHIPPED_PP = PROJECT_ROOT / f"{MODULE_NAME}.pp" diff --git a/tests/resources/ceph/test_requirements_ceph.py b/tests/resources/ceph/test_requirements_ceph.py index 3e93261f..756b5aeb 100644 --- a/tests/resources/ceph/test_requirements_ceph.py +++ b/tests/resources/ceph/test_requirements_ceph.py @@ -1,4 +1,6 @@ +import contextlib import hashlib +import subprocess import pytest from unittest.mock import AsyncMock, MagicMock, patch @@ -9,6 +11,8 @@ HasSudo, LoopControlDeviceExists, LoopControlDeviceWriteable, + BlockPoolDiskGroup, + BlockPoolParentAccessible, SELinuxModule, ) @@ -77,55 +81,25 @@ async def test_loop_control_writeable_check_true(self): result = await self.req.check() assert result is True - async def test_loop_control_writeable_check_false_local(self): - mock_check_proc = AsyncMock() - mock_check_proc.wait = AsyncMock(return_value=1) - - mock_stat_proc = MagicMock() - mock_stat_proc.communicate = MagicMock(return_value=(b"disk", 0)) - - mock_whoami_proc = MagicMock() - mock_whoami_proc.communicate = MagicMock(return_value=(b"testuser", 0)) - - async def side_effect_arun(args): - if "stat" in args: - return mock_stat_proc - if "whoami" in args: - return mock_whoami_proc - return mock_check_proc - - with ( - patch.object(self.req.host, "arun", side_effect=side_effect_arun), - patch.object(self.req.host, "type", "local"), - ): - result = await self.req.check() - assert result is False - assert "usermod" in " ".join(self.req.fix_cmd) - - async def test_loop_control_writeable_check_false_remote(self): - mock_check_proc = AsyncMock() - mock_check_proc.wait = AsyncMock(return_value=1) - - mock_stat_proc = MagicMock() - mock_stat_proc.communicate = MagicMock(return_value=(b"disk", 0)) - - mock_whoami_proc = MagicMock() - mock_whoami_proc.communicate = MagicMock(return_value=(b"testuser", 0)) - - async def side_effect_arun(args): - if "stat" in args: - return mock_stat_proc - if "whoami" in args: - return mock_whoami_proc - return mock_check_proc - + @pytest.mark.parametrize( + "host_type,fix_cmd", + [ + ("local", ["sudo", "usermod", "-a", "-G", "disk", "testuser"]), + ("remote", ["sudo", "chgrp", "testuser", "/dev/loop-control"]), + ], + ) + async def test_loop_control_writeable_check_false(self, host_type, fix_cmd): + mock_check_proc = AsyncMock(wait=AsyncMock(return_value=1)) + mock_stat_proc = MagicMock(communicate=MagicMock(return_value=(b"disk", 0))) + whoami = subprocess.CompletedProcess([], 0, b"testuser\n", b"") with ( - patch.object(self.req.host, "arun", side_effect=side_effect_arun), - patch.object(self.req.host, "type", "remote"), + patch.object(self.req.host, "arun", return_value=mock_check_proc), + patch.object(self.req.host, "run", return_value=mock_stat_proc), + patch.object(self.req.host, "output", return_value=whoami), + patch.object(self.req.host, "type", host_type), ): - result = await self.req.check() - assert result is False - assert "chgrp" in " ".join(self.req.fix_cmd) + assert await self.req.check() is False + assert self.req.fix_cmd == fix_cmd SHIPPED = b"shipped module" @@ -226,55 +200,113 @@ class MockRemoteHost: assert req.fix_cmd[7].endswith("ceph_devstack.pp") -class TestCephDevStackCheckRequirements: - async def test_check_requirements_returns_true_when_all_pass(self): - devstack = CephDevStack() - devstack.service_specs = {} - config["containers"] = {} +class TestBlockPoolRequirements: + @pytest.fixture + def configured_pool(self, tmp_path, pool_parent): + config["block_pool"] = {"parent": pool_parent, "state_dir": str(tmp_path)} + return pool_parent + + @pytest.mark.parametrize("cls", [BlockPoolDiskGroup, BlockPoolParentAccessible]) + async def test_skipped_when_pool_disabled(self, cls): + req = cls() + with patch.object(req.host, "arun") as arun: + assert await req.check() is True + arun.assert_not_called() + + async def test_disk_group_member_passes(self, configured_pool): + req = BlockPoolDiskGroup() + mock_proc = AsyncMock(wait=AsyncMock(return_value=0)) + with patch.object(req.host, "arun", return_value=mock_proc) as arun: + assert await req.check() is True + arun.assert_awaited_once_with(["bash", "-c", "id -nG | grep -qw disk"]) + async def test_disk_group_non_member_gets_fix_cmd(self, configured_pool): + req = BlockPoolDiskGroup() + mock_proc = AsyncMock(wait=AsyncMock(return_value=1)) + whoami = subprocess.CompletedProcess([], 0, b"zack\n", b"") with ( - patch("ceph_devstack.resources.ceph.HasSudo") as MockHasSudo, - patch( - "ceph_devstack.resources.ceph.LoopControlDeviceExists" - ) as MockLoopCtrl, - patch( - "ceph_devstack.resources.ceph.LoopControlDeviceWriteable" - ) as MockLoopCtrlWrite, - patch("ceph_devstack.host.host.selinux_enforcing") as mock_selinux, + patch.object(req.host, "arun", return_value=mock_proc), + patch.object(req.host, "output", return_value=whoami), ): - MockHasSudo.return_value = AsyncMock(evaluate=AsyncMock(return_value=True)) - MockLoopCtrl.return_value = AsyncMock(evaluate=AsyncMock(return_value=True)) - MockLoopCtrlWrite.return_value = AsyncMock( - evaluate=AsyncMock(return_value=True) + assert await req.check() is False + assert req.fix_cmd == [ + "sudo", + "bash", + "-c", + "grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group" + " >> /etc/group; usermod -a -G disk zack", + ] + + @pytest.mark.parametrize("fix", [False, True]) + @pytest.mark.parametrize("cls", [BlockPoolDiskGroup, BlockPoolParentAccessible]) + async def test_invalid_pool_config_fails_without_traceback(self, cls, fix, caplog): + config["block_pool"] = {"parent": "/dev/dm-0"} + req = cls() + with ( + patch.dict(config, {"args": {"fix": fix}}), + patch.object(req.host, "arun") as arun, + ): + assert await req.evaluate() is False + arun.assert_not_called() + assert "not allowed" in caplog.text + + @pytest.mark.parametrize("returncode,expected", [(0, True), (1, False)]) + async def test_parent_access_checked_on_host( + self, configured_pool, returncode, expected + ): + req = BlockPoolParentAccessible() + mock_proc = AsyncMock(wait=AsyncMock(return_value=returncode)) + with patch.object(req.host, "arun", return_value=mock_proc) as arun: + assert await req.check() is expected + arun.assert_awaited_once_with( + ["test", "-r", configured_pool, "-a", "-w", configured_pool] + ) + + +CORE_REQUIREMENTS = [ + "HasSudo", + "LoopControlDeviceExists", + "LoopControlDeviceWriteable", + "BlockPoolDiskGroup", + "BlockPoolParentAccessible", +] + + +@pytest.fixture +def requirement_results(): + """Stub CephDevStack's requirement classes; set an entry False to fail it.""" + results = dict.fromkeys(CORE_REQUIREMENTS, True) + with contextlib.ExitStack() as stack: + for name in CORE_REQUIREMENTS: + mock = stack.enter_context(patch(f"ceph_devstack.resources.ceph.{name}")) + mock.return_value.evaluate = AsyncMock( + side_effect=lambda name=name: results[name] ) - mock_selinux.return_value = False - result = await devstack.check_requirements() - assert result is True + stack.enter_context( + patch("ceph_devstack.host.host.selinux_enforcing", return_value=False) + ) + yield results + - async def test_check_requirements_returns_false_when_repo_missing(self): +class TestCephDevStackCheckRequirements: + @pytest.mark.parametrize("failing", [None, *CORE_REQUIREMENTS]) + async def test_check_requirements_fails_if_any_requirement_fails( + self, requirement_results, failing + ): + devstack = CephDevStack() + devstack.service_specs = {} + config["containers"] = {} + if failing: + requirement_results[failing] = False + assert await devstack.check_requirements() is (failing is None) + + async def test_check_requirements_returns_false_when_repo_missing( + self, requirement_results + ): devstack = CephDevStack() devstack.service_specs = {} config["containers"] = { "custom": {"repo": "/nonexistent/path"}, } - - with ( - patch("ceph_devstack.resources.ceph.HasSudo") as MockHasSudo, - patch( - "ceph_devstack.resources.ceph.LoopControlDeviceExists" - ) as MockLoopCtrl, - patch( - "ceph_devstack.resources.ceph.LoopControlDeviceWriteable" - ) as MockLoopCtrlWrite, - patch("ceph_devstack.host.host.selinux_enforcing") as mock_selinux, - patch("ceph_devstack.host.host.path_exists") as mock_path_exists, - ): - MockHasSudo.return_value = AsyncMock(evaluate=AsyncMock(return_value=True)) - MockLoopCtrl.return_value = AsyncMock(evaluate=AsyncMock(return_value=True)) - MockLoopCtrlWrite.return_value = AsyncMock( - evaluate=AsyncMock(return_value=True) - ) - mock_selinux.return_value = False - mock_path_exists.return_value = False - result = await devstack.check_requirements() - assert result is False + with patch("ceph_devstack.host.host.path_exists", return_value=False): + assert await devstack.check_requirements() is False diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 00000000..e925f59e --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,49 @@ +import sys +from unittest.mock import patch + +import pytest + +from ceph_devstack import config, parse_args +from ceph_devstack.cli import block_pool_status, main + + +class TestBlockPoolCLI: + def test_parse_args_block_pool_status(self): + args = parse_args(["block-pool", "status"]) + assert args.command == "block-pool" + assert args.block_pool_op == "status" + + @pytest.mark.parametrize("argv,rc", [(["status"], 0), ([], 2)]) + def test_main_dispatches_block_pool_ops(self, argv, rc): + with ( + patch.object(sys, "argv", ["ceph-devstack", "block-pool", *argv]), + patch.object(type(config), "load"), + patch("ceph_devstack.cli.block_pool_status", return_value=0) as status, + ): + assert main() == rc + assert status.call_count == (rc == 0) + + def test_status_disabled(self, caplog): + with caplog.at_level("INFO", logger="ceph-devstack"): + assert block_pool_status() == 0 + assert "disabled" in caplog.text + + def test_status_invalid_parent_fails(self, caplog): + config["block_pool"] = {"parent": "/dev/dm-0"} + assert block_pool_status() == 1 + assert "not allowed" in caplog.text + + def test_status_shows_allocations(self, tmp_path, enrolled_pool, caplog): + enrolled_pool.get_or_allocate("ceph_node", 0, 1024**2) + enrolled_pool.get_or_allocate("testnode_0", 0, 1024**2) + enrolled_pool.release_owner("testnode_0") + config["block_pool"] = { + "parent": enrolled_pool.parent, + "state_dir": str(tmp_path), + } + with caplog.at_level("INFO", logger="ceph-devstack"): + assert block_pool_status() == 0 + assert f"pool id: {enrolled_pool.pool_id}" in caplog.text + assert "ceph_node:0: offset 0, size 1M, last owner ceph_node" in caplog.text + assert "offset 1048576, size 1M" in caplog.text + assert "next offset: 2097152" in caplog.text diff --git a/tests/test_config.py b/tests/test_config.py index cd4ed7d9..b399c430 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -104,6 +104,7 @@ class TestConfigDefaults: def test_config_defaults(self): assert config == { "data_dir": "~/.local/share/ceph-devstack", + "block_pool": {}, "containers": { "archive": {"image": "python:alpine"}, "beanstalk": {"image": "quay.io/ceph-infra/teuthology-beanstalkd:main"},