Device Mapper

This section documents the Device Mapper functionality, which provides interfaces for working with Linux's device-mapper framework.

Device Mapper Core

sts.dm

Device Mapper device management.

This module provides functionality for managing Device Mapper devices: - Device discovery - Device information - Device operations - Device Mapper targets

Device Mapper is a Linux kernel framework for mapping physical block devices onto higher-level virtual block devices. It forms the foundation for (example): - LVM (Logical Volume Management) - Software RAID (dm-raid) - Disk encryption (dm-crypt) - Thin provisioning (dm-thin)

DelayTarget dataclass

Bases: DmTarget

Delay target.

Delays I/O operations by a specified amount. Useful for testing how applications handle slow devices.

Args format:

Example

target = DelayTarget(0, 1000000, '253:0 0 100')  # 100ms delay
str(target)
'0 1000000 delay 253:0 0 100'

Source code in sts_libs/src/sts/dm.py

@dataclass
class DelayTarget(DmTarget):
    """Delay target.

    Delays I/O operations by a specified amount. Useful for testing
    how applications handle slow devices.

    Args format: <device> <offset> <delay in milliseconds>

    Example:
        ```python
        target = DelayTarget(0, 1000000, '253:0 0 100')  # 100ms delay
        str(target)
        '0 1000000 delay 253:0 0 100'
        ```
    """

DmDevice dataclass

Bases: StorageDevice

Device Mapper device representation.

A Device Mapper device is a virtual block device that maps to one or more physical devices through a table of targets.

Parameters:

Name	Type	Description	Default
name	str | None	Device name (optional, e.g. 'dm-0')	None
path	Path | str | None	Device path (optional, defaults to /dev/)	None
size	int | None	Device size in bytes (optional, discovered from device)	None
dm_name	str | None	Device Mapper name (optional, discovered from device)	None
model	str | None	Device model (optional)	None
uuid	str | None	Device UUID (optional)	None

Example

device = DmDevice('dm-0')  # Create from kernel name
device = DmDevice(dm_name='vg-lv')  # Create from mapper name

Source code in sts_libs/src/sts/dm.py

@dataclass
class DmDevice(StorageDevice):
    """Device Mapper device representation.

    A Device Mapper device is a virtual block device that maps to one or
    more physical devices through a table of targets.

    Args:
        name: Device name (optional, e.g. 'dm-0')
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        dm_name: Device Mapper name (optional, discovered from device)
        model: Device model (optional)
        uuid: Device UUID (optional)

    Example:
        ```python
        device = DmDevice('dm-0')  # Create from kernel name
        device = DmDevice(dm_name='vg-lv')  # Create from mapper name
        ```
    """

    # Optional parameters from parent classes
    name: str | None = None
    path: Path | str | None = None
    size: int | None = None
    model: str | None = None

    # Optional parameters for this class
    dm_name: str | None = None  # Mapper name (e.g. 'vg-lv')
    uuid: str | None = None  # Unique identifier

    # Internal fields
    _table: str | None = field(init=False, default=None)

    # Class-level paths
    DM_PATH: ClassVar[Path] = Path('/sys/class/block')  # Sysfs block devices
    DM_DEV_PATH: ClassVar[Path] = Path('/dev/mapper')  # Device nodes

    def __post_init__(self) -> None:
        """Initialize Device Mapper device.

        - Sets device path if not provided
        - Discovers mapper name if not provided
        - Loads device table

        Raises:
            DeviceNotFoundError: If device does not exist
            DeviceError: If device cannot be accessed
        """
        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/{self.name}'

        # Initialize parent class
        super().__post_init__()

        # Get device mapper name if not provided
        if not self.dm_name and self.name:
            result = run(f'dmsetup info -c --noheadings -o name {self.name}')
            if result.succeeded:
                self.dm_name = result.stdout.strip()

        # Get device table (defines how I/O is mapped)
        if self.dm_name:
            result = run(f'dmsetup table {self.dm_name}')
            if result.succeeded:
                self._table = result.stdout.strip()

    @property
    def device_path(self) -> Path:
        """Get path to device in sysfs.

        The sysfs path provides access to device attributes and statistics.

        Returns:
            Path to device directory

        Raises:
            DeviceNotFoundError: If device does not exist

        Example:
            ```python
            device.device_path
            PosixPath('/sys/class/block/dm-0')
            ```
        """
        if not self.name:
            msg = 'Device name not available'
            raise DeviceNotFoundError(msg)

        path = self.DM_PATH / self.name
        if not path.exists():
            msg = f'Device {self.name} not found'
            raise DeviceNotFoundError(msg)
        return path

    @property
    def table(self) -> str | None:
        """Get device table.

        The table defines how I/O requests are mapped to underlying devices.
        Format: <start sector> <length> <target type> <target args>

        Returns:
            Device table string or None if not available

        Example:
            ```python
            device.table
            '0 209715200 linear 253:0 0'
            ```
        """
        return self._table

    def suspend(self) -> bool:
        """Suspend device.

        Suspends I/O to the device. Required before changing the device table.
        Outstanding I/O will be queued until the device is resumed.

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.suspend()
            True
            ```
        """
        if not self.dm_name:
            logging.error('Device mapper name not available')
            return False

        result = run(f'dmsetup suspend {self.dm_name}')
        if result.failed:
            logging.error('Failed to suspend device')
            return False
        return True

    def resume(self) -> bool:
        """Resume device.

        Resumes I/O to the device after suspension.
        Queued I/O will be processed.

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.resume()
            True
            ```
        """
        if not self.dm_name:
            logging.error('Device mapper name not available')
            return False

        result = run(f'dmsetup resume {self.dm_name}')
        if result.failed:
            logging.error('Failed to resume device')
            return False
        return True

    def remove(self) -> bool:
        """Remove device.

        Removes the device mapping. The underlying devices are unaffected.
        Device must not be in use (mounted, etc).

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            device.remove()
            True
            ```
        """
        if not self.dm_name:
            logging.error('Device mapper name not available')
            return False

        result = run(f'dmsetup remove {self.dm_name}')
        if result.failed:
            logging.error('Failed to remove device')
            return False
        return True

    @classmethod
    def get_all(cls) -> Sequence[DmDevice]:
        """Get list of all Device Mapper devices.

        Lists all active device-mapper devices on the system.
        Currently only lists linear targets.

        Returns:
            List of DmDevice instances

        Example:
            ```python
            DmDevice.get_all()
            [DmDevice(name='dm-0', ...), DmDevice(name='dm-1', ...)]
            ```
        """
        devices = []
        # List only linear targets for now
        result = run('dmsetup ls --target linear')
        if result.failed:
            logging.warning('No Device Mapper devices found')
            return []

        for line in result.stdout.splitlines():
            try:
                # Parse line like: vg-lv (253:0)
                dm_name, dev_id = line.split()
                dev_id = dev_id.strip('()')
                major, minor = dev_id.split(':')

                # Get kernel device name (dm-N)
                result = run(f'ls -l /dev/dm-* | grep "{major}, *{minor}"')
                if result.failed:
                    continue
                name = result.stdout.split('/')[-1].strip()

                # Get device UUID for identification
                result = run(f'dmsetup info -c --noheadings -o uuid {dm_name}')
                uuid = result.stdout.strip() if result.succeeded else None

                devices.append(cls(name=name, dm_name=dm_name, uuid=uuid))
            except (ValueError, DeviceError):
                logging.exception('Failed to parse device info')
                continue

        return devices

    @classmethod
    def get_by_name(cls, dm_name: str) -> DmDevice | None:
        """Get Device Mapper device by name.

        Finds a device by its mapper name (e.g. 'vg-lv').
        More user-friendly than using kernel names (dm-N).

        Args:
            dm_name: Device Mapper name (e.g. 'vg-lv')

        Returns:
            DmDevice instance or None if not found

        Example:
            ```python
            DmDevice.get_by_name('vg-lv')
            DmDevice(name='dm-0', ...)
            ```
        """
        if not dm_name:
            msg = 'Device Mapper name required'
            raise ValueError(msg)

        for device in cls.get_all():
            if device.dm_name == dm_name:
                return device

        return None

device_path: Path property

Get path to device in sysfs.

The sysfs path provides access to device attributes and statistics.

Returns:

Type	Description
Path	Path to device directory

Raises:

Type	Description
DeviceNotFoundError	If device does not exist

Example

device.device_path
PosixPath('/sys/class/block/dm-0')

table: str | None property

Get device table.

The table defines how I/O requests are mapped to underlying devices. Format:

Returns:

Type	Description
str | None	Device table string or None if not available

Example

device.table
'0 209715200 linear 253:0 0'

__post_init__()

Initialize Device Mapper device.

• Sets device path if not provided
• Discovers mapper name if not provided
• Loads device table

Raises:

Type	Description
DeviceNotFoundError	If device does not exist
DeviceError	If device cannot be accessed

Source code in sts_libs/src/sts/dm.py

def __post_init__(self) -> None:
    """Initialize Device Mapper device.

    - Sets device path if not provided
    - Discovers mapper name if not provided
    - Loads device table

    Raises:
        DeviceNotFoundError: If device does not exist
        DeviceError: If device cannot be accessed
    """
    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/{self.name}'

    # Initialize parent class
    super().__post_init__()

    # Get device mapper name if not provided
    if not self.dm_name and self.name:
        result = run(f'dmsetup info -c --noheadings -o name {self.name}')
        if result.succeeded:
            self.dm_name = result.stdout.strip()

    # Get device table (defines how I/O is mapped)
    if self.dm_name:
        result = run(f'dmsetup table {self.dm_name}')
        if result.succeeded:
            self._table = result.stdout.strip()

get_all() classmethod

Get list of all Device Mapper devices.

Lists all active device-mapper devices on the system. Currently only lists linear targets.

Returns:

Type	Description
Sequence[DmDevice]	List of DmDevice instances

Example

DmDevice.get_all()
[DmDevice(name='dm-0', ...), DmDevice(name='dm-1', ...)]

Source code in sts_libs/src/sts/dm.py

@classmethod
def get_all(cls) -> Sequence[DmDevice]:
    """Get list of all Device Mapper devices.

    Lists all active device-mapper devices on the system.
    Currently only lists linear targets.

    Returns:
        List of DmDevice instances

    Example:
        ```python
        DmDevice.get_all()
        [DmDevice(name='dm-0', ...), DmDevice(name='dm-1', ...)]
        ```
    """
    devices = []
    # List only linear targets for now
    result = run('dmsetup ls --target linear')
    if result.failed:
        logging.warning('No Device Mapper devices found')
        return []

    for line in result.stdout.splitlines():
        try:
            # Parse line like: vg-lv (253:0)
            dm_name, dev_id = line.split()
            dev_id = dev_id.strip('()')
            major, minor = dev_id.split(':')

            # Get kernel device name (dm-N)
            result = run(f'ls -l /dev/dm-* | grep "{major}, *{minor}"')
            if result.failed:
                continue
            name = result.stdout.split('/')[-1].strip()

            # Get device UUID for identification
            result = run(f'dmsetup info -c --noheadings -o uuid {dm_name}')
            uuid = result.stdout.strip() if result.succeeded else None

            devices.append(cls(name=name, dm_name=dm_name, uuid=uuid))
        except (ValueError, DeviceError):
            logging.exception('Failed to parse device info')
            continue

    return devices

get_by_name(dm_name) classmethod

Get Device Mapper device by name.

Finds a device by its mapper name (e.g. 'vg-lv'). More user-friendly than using kernel names (dm-N).

Parameters:

Name	Type	Description	Default
dm_name	str	Device Mapper name (e.g. 'vg-lv')	required

Returns:

Type	Description
DmDevice | None	DmDevice instance or None if not found

Example

DmDevice.get_by_name('vg-lv')
DmDevice(name='dm-0', ...)

Source code in sts_libs/src/sts/dm.py

@classmethod
def get_by_name(cls, dm_name: str) -> DmDevice | None:
    """Get Device Mapper device by name.

    Finds a device by its mapper name (e.g. 'vg-lv').
    More user-friendly than using kernel names (dm-N).

    Args:
        dm_name: Device Mapper name (e.g. 'vg-lv')

    Returns:
        DmDevice instance or None if not found

    Example:
        ```python
        DmDevice.get_by_name('vg-lv')
        DmDevice(name='dm-0', ...)
        ```
    """
    if not dm_name:
        msg = 'Device Mapper name required'
        raise ValueError(msg)

    for device in cls.get_all():
        if device.dm_name == dm_name:
            return device

    return None

remove()

Remove device.

Removes the device mapping. The underlying devices are unaffected. Device must not be in use (mounted, etc).

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.remove()
True

Source code in sts_libs/src/sts/dm.py

def remove(self) -> bool:
    """Remove device.

    Removes the device mapping. The underlying devices are unaffected.
    Device must not be in use (mounted, etc).

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.remove()
        True
        ```
    """
    if not self.dm_name:
        logging.error('Device mapper name not available')
        return False

    result = run(f'dmsetup remove {self.dm_name}')
    if result.failed:
        logging.error('Failed to remove device')
        return False
    return True

resume()

Resume device.

Resumes I/O to the device after suspension. Queued I/O will be processed.

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.resume()
True

Source code in sts_libs/src/sts/dm.py

def resume(self) -> bool:
    """Resume device.

    Resumes I/O to the device after suspension.
    Queued I/O will be processed.

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.resume()
        True
        ```
    """
    if not self.dm_name:
        logging.error('Device mapper name not available')
        return False

    result = run(f'dmsetup resume {self.dm_name}')
    if result.failed:
        logging.error('Failed to resume device')
        return False
    return True

suspend()

Suspend device.

Suspends I/O to the device. Required before changing the device table. Outstanding I/O will be queued until the device is resumed.

Returns:

Type	Description
bool	True if successful, False otherwise

Example

device.suspend()
True

Source code in sts_libs/src/sts/dm.py

def suspend(self) -> bool:
    """Suspend device.

    Suspends I/O to the device. Required before changing the device table.
    Outstanding I/O will be queued until the device is resumed.

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        device.suspend()
        True
        ```
    """
    if not self.dm_name:
        logging.error('Device mapper name not available')
        return False

    result = run(f'dmsetup suspend {self.dm_name}')
    if result.failed:
        logging.error('Failed to suspend device')
        return False
    return True

DmTarget dataclass

Base class for Device Mapper targets.

Device Mapper targets define how I/O operations are processed. Each target maps a range of the virtual device to one or more physical devices, optionally transforming the I/O in some way.

Parameters:

Name	Type	Description	Default
start	int	Start sector (where this target begins)	required
size	int	Size in sectors (length of this target)	required
args	str	Target-specific arguments (e.g. device paths, options)	required

Source code in sts_libs/src/sts/dm.py

@dataclass
class DmTarget:
    """Base class for Device Mapper targets.

    Device Mapper targets define how I/O operations are processed.
    Each target maps a range of the virtual device to one or more
    physical devices, optionally transforming the I/O in some way.

    Args:
        start: Start sector (where this target begins)
        size: Size in sectors (length of this target)
        args: Target-specific arguments (e.g. device paths, options)
    """

    start: int
    size: int
    args: str

    @property
    def type(self) -> str:
        """Get target type.

        Returns:
            Target type string (e.g. 'linear', 'thin-pool')
        """
        # Remove 'target' suffix from class name and convert to lowercase
        return self.__class__.__name__.lower().removesuffix('target')

    def __str__(self) -> str:
        """Return target table entry.

        Format: <start> <size> <type> <args>
        Used in dmsetup table commands.
        """
        return f'{self.start} {self.size} {self.type} {self.args}'

type: str property

Get target type.

Returns:

Type	Description
str	Target type string (e.g. 'linear', 'thin-pool')

__str__()

Return target table entry.

Format: Used in dmsetup table commands.

Source code in sts_libs/src/sts/dm.py

def __str__(self) -> str:
    """Return target table entry.

    Format: <start> <size> <type> <args>
    Used in dmsetup table commands.
    """
    return f'{self.start} {self.size} {self.type} {self.args}'

LinearTarget dataclass

Bases: DmTarget

Linear target.

The simplest target type - maps a linear range of the virtual device directly onto a linear range of another device.

Args format:

Example

target = LinearTarget(0, 1000000, '253:0 0')  # Map to device 253:0 starting at sector 0
str(target)
'0 1000000 linear 253:0 0'

Source code in sts_libs/src/sts/dm.py

@dataclass
class LinearTarget(DmTarget):
    """Linear target.

    The simplest target type - maps a linear range of the virtual device
    directly onto a linear range of another device.

    Args format: <destination device> <sector offset>

    Example:
        ```python
        target = LinearTarget(0, 1000000, '253:0 0')  # Map to device 253:0 starting at sector 0
        str(target)
        '0 1000000 linear 253:0 0'
        ```
    """

MultipathTarget dataclass

Bases: DmTarget

Multipath target.

Provides I/O failover across multiple paths to the same device. Used for high availability storage configurations.

Args format: <# of feature args> <# of paths>

Example

# Round-robin across two paths with queue_if_no_path
target = MultipathTarget(0, 1000000, '2 1 round-robin 0 2 1 8:32 1000 8:48 1000')
str(target)
'0 1000000 multipath 2 1 round-robin 0 2 1 8:32 1000 8:48 1000'

Source code in sts_libs/src/sts/dm.py

@dataclass
class MultipathTarget(DmTarget):
    """Multipath target.

    Provides I/O failover across multiple paths to the same device.
    Used for high availability storage configurations.

    Args format: <# of feature args> <# of paths> <features...> <path specs...>

    Example:
        ```python
        # Round-robin across two paths with queue_if_no_path
        target = MultipathTarget(0, 1000000, '2 1 round-robin 0 2 1 8:32 1000 8:48 1000')
        str(target)
        '0 1000000 multipath 2 1 round-robin 0 2 1 8:32 1000 8:48 1000'
        ```
    """

ThinPoolTarget dataclass

Bases: DmTarget

Thin pool target.

Manages a pool of storage space from which thin volumes can be allocated. Enables thin provisioning - allocating more virtual space than physical.

Args format:

Example

# Pool with 128 sector blocks and 32768 low water mark
target = ThinPoolTarget(0, 1000000, '253:0 253:1 128 32768 1 skip_block_zeroing')
str(target)
'0 1000000 thin-pool 253:0 253:1 128 32768 1 skip_block_zeroing'

Source code in sts_libs/src/sts/dm.py

@dataclass
class ThinPoolTarget(DmTarget):
    """Thin pool target.

    Manages a pool of storage space from which thin volumes can be allocated.
    Enables thin provisioning - allocating more virtual space than physical.

    Args format: <metadata dev> <data dev> <block size> <low water mark> <flags> <args>

    Example:
        ```python
        # Pool with 128 sector blocks and 32768 low water mark
        target = ThinPoolTarget(0, 1000000, '253:0 253:1 128 32768 1 skip_block_zeroing')
        str(target)
        '0 1000000 thin-pool 253:0 253:1 128 32768 1 skip_block_zeroing'
        ```
    """

ThinTarget dataclass

Bases: DmTarget

Thin target.

A virtual device that allocates space from a thin pool on demand. Enables over-provisioning of storage.

Args format:

Example

target = ThinTarget(0, 1000000, '253:0 1')  # Device 1 from pool 253:0
str(target)
'0 1000000 thin 253:0 1'

Source code in sts_libs/src/sts/dm.py

@dataclass
class ThinTarget(DmTarget):
    """Thin target.

    A virtual device that allocates space from a thin pool on demand.
    Enables over-provisioning of storage.

    Args format: <pool dev> <dev id>

    Example:
        ```python
        target = ThinTarget(0, 1000000, '253:0 1')  # Device 1 from pool 253:0
        str(target)
        '0 1000000 thin 253:0 1'
        ```
    """

Device Mapper Persistent Data

sts.dmpd

Device mapper persistent data tools.

This module provides functionality for device-mapper-persistent-data tools: - Cache tools (cache_check, cache_dump, etc.) - Thin provisioning tools (thin_check, thin_dump, etc.) - Metadata repair and restore

Device Mapper Persistent Data: - Manages metadata for advanced DM targets - Ensures data consistency across reboots - Provides tools for metadata maintenance - Supports metadata backup and restore

DeviceMapperPD

Device mapper persistent data tools.

Provides tools for managing metadata: - Cache metadata (dm-cache) - Thin pool metadata (dm-thin) - Metadata validation - Metadata repair - Backup/restore

Key operations: 1. Validation (check) 2. Backup (dump) 3. Repair (repair) 4. Restore (restore)

Source code in sts_libs/src/sts/dmpd.py

class DeviceMapperPD:
    """Device mapper persistent data tools.

    Provides tools for managing metadata:
    - Cache metadata (dm-cache)
    - Thin pool metadata (dm-thin)
    - Metadata validation
    - Metadata repair
    - Backup/restore

    Key operations:
    1. Validation (check)
    2. Backup (dump)
    3. Repair (repair)
    4. Restore (restore)
    """

    # Available commands and their valid arguments
    # Each command supports specific subset of options
    COMMANDS: ClassVar[dict[str, set[str]]] = {
        'cache_check': {'quiet', 'super_block_only', 'skip_mappings'},
        'cache_dump': {'output', 'repair', 'format'},
        'cache_repair': {'input', 'output'},
        'cache_restore': {'input', 'output', 'quiet'},
        'thin_check': {'quiet', 'super_block_only', 'skip_mappings'},
        'thin_dump': {'output', 'repair', 'format', 'metadata_snap'},
        'thin_repair': {'input', 'output'},
        'thin_restore': {'input', 'output', 'quiet'},
    }

    def __init__(self) -> None:
        """Initialize device mapper tools.

        Ensures required tools are installed:
        - device-mapper-persistent-data package
        - Cache tools
        - Thin provisioning tools
        """
        # Install required package
        pm = Dnf()
        if not pm.install('device-mapper-persistent-data'):
            msg = 'Failed to install device-mapper-persistent-data'
            raise RuntimeError(msg)

    def _convert_options(self, options: Mapping[str, Any] | None) -> dict[str, str]:
        """Convert options to command arguments.

        Converts Python options to CLI format:
        - Filters None values
        - Converts values to strings
        - Preserves option order

        Args:
            options: Command options

        Returns:
            Command arguments
        """
        if not options:
            return {}
        return {k: str(v) for k, v in options.items() if v is not None}

    def _run(self, cmd: str, valid_args: set[str], options: Mapping[str, Any] | None = None) -> bool:
        """Run command with arguments.

        Executes command with validation:
        - Filters invalid arguments
        - Formats command string
        - Captures output and errors

        Args:
            cmd: Command to run
            valid_args: Valid arguments for command
            options: Command options

        Returns:
            True if successful, False otherwise
        """
        # Filter valid arguments
        args = {k: v for k, v in self._convert_options(options).items() if k in valid_args}

        # Build and run command
        cmd = f'{cmd} {format_args(**args)}'
        result = run(cmd)
        if result.failed:
            logging.error(f'Command failed: {cmd}')
            return False
        return True

    def _check_metadata(self, cmd: str, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Check metadata.

        Common metadata validation:
        - Validates source device
        - Runs appropriate check command
        - Handles command options

        Args:
            cmd: Command to run
            source: Source device
            options: Command options

        Returns:
            True if check passed, False otherwise
        """
        device = source.get_path()
        if not device:
            logging.error('Either source_file or source_vg/source_lv required')
            return False

        return self._run(f'{cmd} {device}', self.COMMANDS[cmd], options)

    def cache_check(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Check cache metadata.

        Validates dm-cache metadata:
        - Superblock integrity
        - Mapping correctness
        - Reference counts
        - Free space

        Args:
            source: Source device
            options: Command options

        Returns:
            True if check passed, False otherwise

        Example:
            ```python
            dmpd = DeviceMapperPD()
            source = DeviceSource(vg='vg0', lv='cache0')
            dmpd.cache_check(source, {'quiet': None})
            True
            ```
        """
        return self._check_metadata('cache_check', source, options)

    def thin_check(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Check thin metadata.

        Validates dm-thin metadata:
        - Superblock integrity
        - Device mappings
        - Space maps
        - Reference counts

        Args:
            source: Source device
            options: Command options

        Returns:
            True if check passed, False otherwise

        Example:
            ```python
            dmpd = DeviceMapperPD()
            source = DeviceSource(vg='vg0', lv='thin0')
            dmpd.thin_check(source, {'quiet': None})
            True
            ```
        """
        return self._check_metadata('thin_check', source, options)

    def thin_dump(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Dump thin metadata.

        Creates metadata backup:
        - XML or human readable format
        - Optional repair during dump
        - Can use metadata snapshot

        Args:
            source: Source device
            options: Command options

        Returns:
            True if dump succeeded, False otherwise

        Example:
            ```python
            dmpd = DeviceMapperPD()
            source = DeviceSource(vg='vg0', lv='thin0')
            dmpd.thin_dump(source, {'output': 'metadata.xml'})
            True
            ```
        """
        return self._check_metadata('thin_dump', source, options)

    def thin_restore(self, source_file: str | Path, target: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Restore thin metadata.

        Restores metadata from backup:
        - Validates input file
        - Creates new metadata
        - Preserves device mappings

        Args:
            source_file: Source metadata file
            target: Target device
            options: Command options

        Returns:
            True if restore succeeded, False otherwise

        Example:
            ```python
            dmpd = DeviceMapperPD()
            target = DeviceSource(vg='vg0', lv='thin0')
            dmpd.thin_restore('metadata.xml', target)
            True
            ```
        """
        source_path = Path(source_file)
        if not source_path.is_file():
            logging.error('Source file does not exist')
            return False

        target_path = target.get_path()
        if not target_path:
            logging.error('Either target_file or target_vg/target_lv required')
            return False

        restore_options = cast(dict[str, Any], {'input': str(source_path), 'output': target_path})
        if options:
            restore_options.update(cast(dict[str, Any], options))

        return self._run('thin_restore', self.COMMANDS['thin_restore'], restore_options)

    def thin_repair(self, source: DeviceSource, target: DeviceSource, options: DmpdOptions | None = None) -> bool:
        """Repair thin metadata.

        Attempts to repair corrupted metadata:
        - Reads corrupted metadata
        - Fixes inconsistencies
        - Creates repaired copy

        Args:
            source: Source device (corrupted)
            target: Target device (repaired)
            options: Command options

        Returns:
            True if repair succeeded, False otherwise

        Example:
            ```python
            dmpd = DeviceMapperPD()
            source = DeviceSource(vg='vg0', lv='thin0')
            target = DeviceSource(file='repaired.xml')
            dmpd.thin_repair(source, target)
            True
            ```
        """
        source_path = source.get_path()
        if not source_path:
            logging.error('Either source_file or source_vg/source_lv required')
            return False

        target_path = target.get_path()
        if not target_path:
            logging.error('Either target_file or target_vg/target_lv required')
            return False

        repair_options = cast(dict[str, Any], {'input': source_path, 'output': target_path})
        if options:
            repair_options.update(cast(dict[str, Any], options))

        return self._run('thin_repair', self.COMMANDS['thin_repair'], repair_options)

__init__()

Initialize device mapper tools.

Ensures required tools are installed: - device-mapper-persistent-data package - Cache tools - Thin provisioning tools

Source code in sts_libs/src/sts/dmpd.py

def __init__(self) -> None:
    """Initialize device mapper tools.

    Ensures required tools are installed:
    - device-mapper-persistent-data package
    - Cache tools
    - Thin provisioning tools
    """
    # Install required package
    pm = Dnf()
    if not pm.install('device-mapper-persistent-data'):
        msg = 'Failed to install device-mapper-persistent-data'
        raise RuntimeError(msg)

cache_check(source, options=None)

Check cache metadata.

Validates dm-cache metadata: - Superblock integrity - Mapping correctness - Reference counts - Free space

Parameters:

Name	Type	Description	Default
source	DeviceSource	Source device	required
options	DmpdOptions | None	Command options	None

Returns:

Type	Description
bool	True if check passed, False otherwise

Example

dmpd = DeviceMapperPD()
source = DeviceSource(vg='vg0', lv='cache0')
dmpd.cache_check(source, {'quiet': None})
True

Source code in sts_libs/src/sts/dmpd.py

def cache_check(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
    """Check cache metadata.

    Validates dm-cache metadata:
    - Superblock integrity
    - Mapping correctness
    - Reference counts
    - Free space

    Args:
        source: Source device
        options: Command options

    Returns:
        True if check passed, False otherwise

    Example:
        ```python
        dmpd = DeviceMapperPD()
        source = DeviceSource(vg='vg0', lv='cache0')
        dmpd.cache_check(source, {'quiet': None})
        True
        ```
    """
    return self._check_metadata('cache_check', source, options)

thin_check(source, options=None)

Check thin metadata.

Validates dm-thin metadata: - Superblock integrity - Device mappings - Space maps - Reference counts

Parameters:

Name	Type	Description	Default
source	DeviceSource	Source device	required
options	DmpdOptions | None	Command options	None

Returns:

Type	Description
bool	True if check passed, False otherwise

Example

dmpd = DeviceMapperPD()
source = DeviceSource(vg='vg0', lv='thin0')
dmpd.thin_check(source, {'quiet': None})
True

Source code in sts_libs/src/sts/dmpd.py

def thin_check(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
    """Check thin metadata.

    Validates dm-thin metadata:
    - Superblock integrity
    - Device mappings
    - Space maps
    - Reference counts

    Args:
        source: Source device
        options: Command options

    Returns:
        True if check passed, False otherwise

    Example:
        ```python
        dmpd = DeviceMapperPD()
        source = DeviceSource(vg='vg0', lv='thin0')
        dmpd.thin_check(source, {'quiet': None})
        True
        ```
    """
    return self._check_metadata('thin_check', source, options)

thin_dump(source, options=None)

Dump thin metadata.

Creates metadata backup: - XML or human readable format - Optional repair during dump - Can use metadata snapshot

Parameters:

Name	Type	Description	Default
source	DeviceSource	Source device	required
options	DmpdOptions | None	Command options	None

Returns:

Type	Description
bool	True if dump succeeded, False otherwise

Example

dmpd = DeviceMapperPD()
source = DeviceSource(vg='vg0', lv='thin0')
dmpd.thin_dump(source, {'output': 'metadata.xml'})
True

Source code in sts_libs/src/sts/dmpd.py

def thin_dump(self, source: DeviceSource, options: DmpdOptions | None = None) -> bool:
    """Dump thin metadata.

    Creates metadata backup:
    - XML or human readable format
    - Optional repair during dump
    - Can use metadata snapshot

    Args:
        source: Source device
        options: Command options

    Returns:
        True if dump succeeded, False otherwise

    Example:
        ```python
        dmpd = DeviceMapperPD()
        source = DeviceSource(vg='vg0', lv='thin0')
        dmpd.thin_dump(source, {'output': 'metadata.xml'})
        True
        ```
    """
    return self._check_metadata('thin_dump', source, options)

thin_repair(source, target, options=None)

Repair thin metadata.

Attempts to repair corrupted metadata: - Reads corrupted metadata - Fixes inconsistencies - Creates repaired copy

Parameters:

Name	Type	Description	Default
source	DeviceSource	Source device (corrupted)	required
target	DeviceSource	Target device (repaired)	required
options	DmpdOptions | None	Command options	None

Returns:

Type	Description
bool	True if repair succeeded, False otherwise

Example

dmpd = DeviceMapperPD()
source = DeviceSource(vg='vg0', lv='thin0')
target = DeviceSource(file='repaired.xml')
dmpd.thin_repair(source, target)
True

Source code in sts_libs/src/sts/dmpd.py

def thin_repair(self, source: DeviceSource, target: DeviceSource, options: DmpdOptions | None = None) -> bool:
    """Repair thin metadata.

    Attempts to repair corrupted metadata:
    - Reads corrupted metadata
    - Fixes inconsistencies
    - Creates repaired copy

    Args:
        source: Source device (corrupted)
        target: Target device (repaired)
        options: Command options

    Returns:
        True if repair succeeded, False otherwise

    Example:
        ```python
        dmpd = DeviceMapperPD()
        source = DeviceSource(vg='vg0', lv='thin0')
        target = DeviceSource(file='repaired.xml')
        dmpd.thin_repair(source, target)
        True
        ```
    """
    source_path = source.get_path()
    if not source_path:
        logging.error('Either source_file or source_vg/source_lv required')
        return False

    target_path = target.get_path()
    if not target_path:
        logging.error('Either target_file or target_vg/target_lv required')
        return False

    repair_options = cast(dict[str, Any], {'input': source_path, 'output': target_path})
    if options:
        repair_options.update(cast(dict[str, Any], options))

    return self._run('thin_repair', self.COMMANDS['thin_repair'], repair_options)

thin_restore(source_file, target, options=None)

Restore thin metadata.

Restores metadata from backup: - Validates input file - Creates new metadata - Preserves device mappings

Parameters:

Name	Type	Description	Default
source_file	str | Path	Source metadata file	required
target	DeviceSource	Target device	required
options	DmpdOptions | None	Command options	None

Returns:

Type	Description
bool	True if restore succeeded, False otherwise

Example

dmpd = DeviceMapperPD()
target = DeviceSource(vg='vg0', lv='thin0')
dmpd.thin_restore('metadata.xml', target)
True

Source code in sts_libs/src/sts/dmpd.py

def thin_restore(self, source_file: str | Path, target: DeviceSource, options: DmpdOptions | None = None) -> bool:
    """Restore thin metadata.

    Restores metadata from backup:
    - Validates input file
    - Creates new metadata
    - Preserves device mappings

    Args:
        source_file: Source metadata file
        target: Target device
        options: Command options

    Returns:
        True if restore succeeded, False otherwise

    Example:
        ```python
        dmpd = DeviceMapperPD()
        target = DeviceSource(vg='vg0', lv='thin0')
        dmpd.thin_restore('metadata.xml', target)
        True
        ```
    """
    source_path = Path(source_file)
    if not source_path.is_file():
        logging.error('Source file does not exist')
        return False

    target_path = target.get_path()
    if not target_path:
        logging.error('Either target_file or target_vg/target_lv required')
        return False

    restore_options = cast(dict[str, Any], {'input': str(source_path), 'output': target_path})
    if options:
        restore_options.update(cast(dict[str, Any], options))

    return self._run('thin_restore', self.COMMANDS['thin_restore'], restore_options)

DeviceSource dataclass

Device source configuration.

Represents a device that can be: - A regular file (metadata dump) - A logical volume (active device) - A device mapper device

Parameters:

Name	Type	Description	Default
file	str | None	Source file path (for metadata files)	None
vg	str | None	Volume group name (for logical volumes)	None
lv	str | None	Logical volume name (for logical volumes)	None

Source code in sts_libs/src/sts/dmpd.py

@dataclass
class DeviceSource:
    """Device source configuration.

    Represents a device that can be:
    - A regular file (metadata dump)
    - A logical volume (active device)
    - A device mapper device

    Args:
        file: Source file path (for metadata files)
        vg: Volume group name (for logical volumes)
        lv: Logical volume name (for logical volumes)
    """

    file: str | None = None
    vg: str | None = None
    lv: str | None = None

    def get_path(self) -> str | None:
        """Get device path.

        Returns path based on configuration:
        - File path if file is specified
        - Device mapper path if VG/LV specified
        - None if invalid configuration

        Returns:
            Device path or None if invalid configuration

        Example:
            ```python
            source = DeviceSource(file='/path/to/file')
            source.get_path()
            '/path/to/file'
            source = DeviceSource(vg='vg0', lv='lv0')
            source.get_path()
            '/dev/mapper/vg0-lv0'
            ```
        """
        if self.file:
            return self.file
        if self.vg and self.lv:
            return get_device_path(self.vg, self.lv)
        return None

get_path()

Get device path.

Returns path based on configuration: - File path if file is specified - Device mapper path if VG/LV specified - None if invalid configuration

Returns:

Type	Description
str | None	Device path or None if invalid configuration

Example

source = DeviceSource(file='/path/to/file')
source.get_path()
'/path/to/file'
source = DeviceSource(vg='vg0', lv='lv0')
source.get_path()
'/dev/mapper/vg0-lv0'

Source code in sts_libs/src/sts/dmpd.py

def get_path(self) -> str | None:
    """Get device path.

    Returns path based on configuration:
    - File path if file is specified
    - Device mapper path if VG/LV specified
    - None if invalid configuration

    Returns:
        Device path or None if invalid configuration

    Example:
        ```python
        source = DeviceSource(file='/path/to/file')
        source.get_path()
        '/path/to/file'
        source = DeviceSource(vg='vg0', lv='lv0')
        source.get_path()
        '/dev/mapper/vg0-lv0'
        ```
    """
    if self.file:
        return self.file
    if self.vg and self.lv:
        return get_device_path(self.vg, self.lv)
    return None

DmpdOptions

Bases: TypedDict

Device mapper persistent data tool options.

Common options for metadata tools: - quiet: Suppress output messages - super_block_only: Only check superblock integrity - skip_mappings: Skip block mapping verification - output: Output file for dumps/repairs - input: Input file for restores/repairs - repair: Attempt metadata repair - format: Output format (xml, human readable) - metadata_snap: Use metadata snapshot for consistency

Source code in sts_libs/src/sts/dmpd.py

class DmpdOptions(TypedDict, total=False):
    """Device mapper persistent data tool options.

    Common options for metadata tools:
    - quiet: Suppress output messages
    - super_block_only: Only check superblock integrity
    - skip_mappings: Skip block mapping verification
    - output: Output file for dumps/repairs
    - input: Input file for restores/repairs
    - repair: Attempt metadata repair
    - format: Output format (xml, human readable)
    - metadata_snap: Use metadata snapshot for consistency
    """

    quiet: str | None
    super_block_only: str | None
    skip_mappings: str | None
    output: str
    input: str
    repair: str | None
    format: str
    metadata_snap: str | None

get_device_path(vg_name, lv_name)

Get device mapper path.

Constructs device mapper path from VG/LV names: /dev/mapper/-

Parameters:

Name	Type	Description	Default
vg_name	str	Volume group name	required
lv_name	str	Logical volume name	required

Returns:

Type	Description
str	Device mapper path

Source code in sts_libs/src/sts/dmpd.py

def get_device_path(vg_name: str, lv_name: str) -> str:
    """Get device mapper path.

    Constructs device mapper path from VG/LV names:
    /dev/mapper/<vg_name>-<lv_name>

    Args:
        vg_name: Volume group name
        lv_name: Logical volume name

    Returns:
        Device mapper path
    """
    return f'/dev/mapper/{vg_name}-{lv_name}'