Device Mapper

Linux kernel framework for mapping physical block devices onto virtual block devices. Underpins LVM, dm-crypt, dm-raid, and dm-multipath. Each device type below wraps dmsetup commands and can exist in two states: configuration-only (has target parameters but no kernel device) or active (created on the system with a /dev/mapper/ path).

DmDevice (base — dmsetup operations)
├── LinearDevice     — 1:1 sector mapping
├── DelayDevice      — adds read/write/flush latency
├── FlakeyDevice     — simulates transient I/O failures
├── CacheDevice      — dm-cache (SSD-backed caching)
├── ThinPoolDevice   — thin provisioning pool
├── ThinDevice       — thin-provisioned volume
├── VdoDevice        — deduplication + compression
├── ZeroDevice       — returns zeros on read, discards writes
└── ErrorDevice      — fails all I/O

Device Mapper Core

sts.dm.base

Base class for Device Mapper devices and dmsetup wrapper methods.

Class Hierarchy::

BlockDevice
    └── DmDevice
            ├── LinearDevice, DelayDevice, VdoDevice
            ├── ThinPoolDevice, ThinDevice
            ├── ZeroDevice, ErrorDevice, FlakeyDevice, CacheDevice
            └── MultipathTarget

Each device can be in two states: configuration (target args only) or active (created on the system with path, dm_name, table, etc.).

For multipath devices managed by multipathd, use sts.multipath.MultipathDevice (inherits BlockDevice directly).

DmDevice pydantic-model

Bases: BlockDevice

Base class for all Device Mapper devices.

Before create(), the device is just a target configuration (start, size_sectors, args). After create(), it becomes a full block device with dm_name, path, table, and all BlockDevice functionality.

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Base class for all Device Mapper devices.\n\nBefore ``create()``, the device is just a target configuration\n(start, size_sectors, args). After ``create()``, it becomes a full\nblock device with dm_name, path, table, and all BlockDevice functionality.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "DmDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• target_type (str)
• table (str | None)
• is_created (bool)

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

class DmDevice(BlockDevice):
    """Base class for all Device Mapper devices.

    Before ``create()``, the device is just a target configuration
    (start, size_sectors, args). After ``create()``, it becomes a full
    block device with dm_name, path, table, and all BlockDevice functionality.
    """

    # Target configuration (always available)
    start: int = 0
    size_sectors: int = 0  # Size in sectors (renamed to avoid conflict with BlockDevice.size)
    args: str = ''

    # Device Mapper name (set after creation or when loading existing device)
    dm_name: str | None = None

    # Target type - override in subclasses
    target_type: str = Field(default='', init=False)

    # Internal state
    table: str | None = Field(default=None, init=False, repr=False)
    is_created: bool = Field(default=False, init=False, repr=False)

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

    def _run(self, cmd: str) -> CommandResult:
        """Run a dmsetup command."""
        return run(cmd)

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive path/name fields only — no I/O."""
        if self.dm_name and not self.path:
            self.path = Path(f'/dev/mapper/{self.dm_name}')
        elif self.name and not self.path:
            self.path = Path(f'/dev/{self.name}')
        if self.path and not self.name:
            self.name = Path(self.path).name
        if self.args:
            self._parse_table()
        return self

    def discover(self) -> Self:
        """Load device state from the system.

        Call after construction for devices that already exist (identified by
        dm_name, name, or path). No-op for configuration-only devices.
        """
        if not self.path and not self.dm_name and not self.name:
            return self

        super().discover()

        # 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()

        # Load table from system
        if self.dm_name:
            result = run(f'dmsetup table {self.dm_name}')
            if result.succeeded:
                self.table = result.stdout.strip()
                self._parse_table()

        self.is_created = True
        return self

    @property
    def type(self) -> str:
        """Target type string (e.g. 'linear', 'delay', 'vdo')."""
        if self.target_type:
            return self.target_type
        # Fallback: Remove 'Device' suffix from class name and convert to lowercase
        return self.__class__.__name__.lower().removesuffix('device')

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

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

    @property
    def device_path(self) -> Path:
        """Path to device in sysfs (``/sys/class/block/<name>``)."""
        if not self.is_created or not self.name:
            msg = 'Device not created yet'
            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 device_root_dir(self) -> str:
        """Device mapper root directory (``/dev/mapper``)."""
        return str(self.DM_DEV_PATH)

    @property
    def dm_device_path(self) -> str | None:
        """Full device path (e.g. ``/dev/mapper/my-device``), or None if dm_name not set."""
        if not self.dm_name:
            return None
        return f'{self.device_root_dir}/{self.dm_name}'

    def create(
        self,
        dm_name: str,
        *,
        uuid: str | None = None,
        readonly: bool = False,
        notable: bool = False,
        readahead: str | int | None = None,
        addnodeoncreate: bool = False,
        addnodeonresume: bool = False,
    ) -> CommandResult:
        """Create the device on the system.

        After successful creation, this device becomes a full BlockDevice
        with all properties populated from the system.

        Args:
            dm_name: Device mapper name
            uuid: UUID for the device (can be used in place of name in commands)
            readonly: Load the table as read-only
            notable: Create device without loading any table
            readahead: Sectors, or 'auto', 'none', or '+N' for minimum
            addnodeoncreate: Ensure /dev/mapper node exists after create
            addnodeonresume: Ensure /dev/mapper node exists after resume
        """
        if self.is_created:
            logger.warning(f'Device {self.dm_name} already created')
            return CommandResult(command=f'dmsetup create {dm_name}', rc=0, stdout='Already created')

        cmd = f'dmsetup create {dm_name}'

        if uuid:
            cmd += f' --uuid {uuid}'
        if readonly:
            cmd += ' --readonly'
        if addnodeoncreate:
            cmd += ' --addnodeoncreate'
        if addnodeonresume:
            cmd += ' --addnodeonresume'
        if readahead is not None:
            cmd += f' --readahead {readahead}'

        if notable:
            cmd += ' --notable'
        else:
            # Build table from this device's configuration
            table = str(self)
            cmd += f' --table "{table}"'
            logger.info(f'Creating device {dm_name} with table: {table}')

        result = self._run(cmd)

        if result.failed:
            logger.error(f'Failed to create device {dm_name}: {result.stderr}')
            return result

        logger.debug(f'Successfully created device mapper device: {dm_name}')

        # Set dm_name and reinitialize as existing device
        self.dm_name = dm_name
        self.path = Path(f'/dev/mapper/{dm_name}')
        self.name = None

        # Get kernel device name
        info_result = run(f'dmsetup info -c --noheadings -o name,blkdevname {dm_name}')
        if info_result.succeeded:
            parts = info_result.stdout.strip().split()
            if len(parts) >= 2:
                self.name = parts[1]

        # Load device state from system (BlockDevice properties, table)
        self.discover()
        return result

    def _parse_table(self) -> None:
        """Parse table and update target configuration from system.

        Subclasses should override this to parse target-specific attributes.
        """
        if not self.table:
            return

        parts = self.table.strip().split(None, 3)
        if len(parts) >= 4:
            self.start = int(parts[0])
            self.size_sectors = int(parts[1])
            # parts[2] is the target type
            self.args = parts[3]

    def refresh(self) -> None:
        """Refresh device table from the system."""
        if not self.is_created or not self.dm_name:
            return

        # Refresh table from system
        result = run(f'dmsetup table {self.dm_name}')
        if result.succeeded:
            self.table = result.stdout.strip()
            self._parse_table()

    def get_status(
        self,
        *,
        target_type: str | None = None,
        noflush: bool = False,
    ) -> str | None:
        """Get device status from dmsetup.

        Args:
            target_type: Only display information for the specified target type
            noflush: Do not commit thin-pool metadata before reporting
        """
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        cmd = f'dmsetup status {self.dm_name}'
        if target_type:
            cmd += f' --target {target_type}'
        if noflush:
            cmd += ' --noflush'

        result = run(cmd)
        if result.failed:
            logger.error(f'Failed to get status for {self.dm_name}')
            return None

        return result.stdout.strip()

    def suspend(
        self,
        *,
        nolockfs: bool = False,
        noflush: bool = False,
    ) -> CommandResult:
        """Suspend I/O to the device.

        Pending I/O is flushed; new I/O is postponed until ``resume()``.
        """
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        cmd = f'dmsetup suspend {self.dm_name}'
        if nolockfs:
            cmd += ' --nolockfs'
        if noflush:
            cmd += ' --noflush'

        result = self._run(cmd)
        if result.failed:
            logger.error('Failed to suspend device')
            return result
        return result

    def resume(
        self,
        *,
        addnodeoncreate: bool = False,
        addnodeonresume: bool = False,
        noflush: bool = False,
        nolockfs: bool = False,
        readahead: str | int | None = None,
    ) -> CommandResult:
        """Resume a suspended device.

        If an inactive table has been loaded, it becomes live.
        """
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        cmd = f'dmsetup resume {self.dm_name}'
        if addnodeoncreate:
            cmd += ' --addnodeoncreate'
        if addnodeonresume:
            cmd += ' --addnodeonresume'
        if noflush:
            cmd += ' --noflush'
        if nolockfs:
            cmd += ' --nolockfs'
        if readahead is not None:
            cmd += f' --readahead {readahead}'

        result = self._run(cmd)
        if result.failed:
            logger.error('Failed to resume device')
            return result
        return result

    def remove(
        self,
        *,
        force: bool = False,
        retry: bool = False,
        deferred: bool = False,
    ) -> CommandResult:
        """Remove the device mapping.

        Args:
            force: Replace the table with one that fails all I/O
            retry: Retry removal for a few seconds if it fails (e.g., udev race)
            deferred: Device removed when last user closes it
        """
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        cmd = f'dmsetup remove {self.dm_name}'
        if force:
            cmd += ' --force'
        if retry:
            cmd += ' --retry'
        if deferred:
            cmd += ' --deferred'

        result = self._run(cmd)
        if result.failed:
            logger.error('Failed to remove device')
            return result

        self.is_created = False
        return result

    def clear(self) -> CommandResult:
        """Destroy the table in the inactive table slot."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup clear {self.dm_name}')
        if result.failed:
            logger.error(f'Failed to clear inactive table for {self.dm_name}')
            return result
        return result

    def deps(self, *, output_format: str = 'devno') -> list[str]:
        """Get devices referenced by the live table.

        Args:
            output_format: 'devno' (major:minor), 'blkdevname', or 'devname'
        """
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return []

        result = run(f'dmsetup deps -o {output_format} {self.dm_name}')
        if result.failed:
            logger.error(f'Failed to get dependencies for {self.dm_name}')
            return []

        # Parse output: "1 dependencies  : (major, minor) ..."
        output = result.stdout.strip()
        deps = []
        if ':' in output:
            deps_part = output.split(':', 1)[1].strip()
            # Extract device identifiers from the output
            if output_format == 'devno':
                # Format: (major, minor) or (major:minor)
                matches = re.findall(r'\((\d+[,:]\s*\d+)\)', deps_part)
                deps = [m.replace(' ', '').replace(',', ':') for m in matches]
            else:
                # Format: (device_name)
                matches = re.findall(r'\(([^)]+)\)', deps_part)
                deps = matches

        return deps

    def info(
        self,
        *,
        columns: bool = False,
        noheadings: bool = False,
        fields: str | None = None,
        separator: str | None = None,
        sort_fields: str | None = None,
        nameprefixes: bool = False,
    ) -> dict[str, str] | str | None:
        """Get device information.

        Returns dict of field->value when columns=False, raw string when
        columns=True, or None on error.
        """
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        cmd = f'dmsetup info {self.dm_name}'
        if columns:
            cmd += ' --columns'
            if noheadings:
                cmd += ' --noheadings'
            if fields:
                cmd += f' -o {fields}'
            if separator:
                cmd += f' --separator "{separator}"'
            if sort_fields:
                cmd += f' --sort {sort_fields}'
            if nameprefixes:
                cmd += ' --nameprefixes'

        result = run(cmd)
        if result.failed:
            logger.error(f'Failed to get info for {self.dm_name}')
            return None

        output = result.stdout.strip()

        if columns:
            return output

        # Parse Field: Value format into dictionary
        info_dict: dict[str, str] = {}
        for line in output.splitlines():
            if ':' in line:
                key, value = line.split(':', 1)
                info_dict[key.strip()] = value.strip()

        return info_dict

    def message(self, sector: int, message: str) -> CommandResult:
        """Send a message to the target at the given sector (use 0 if irrelevant)."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup message {self.dm_name} {sector} {message}')
        if result.failed:
            logger.error(f'Failed to send message to {self.dm_name}: {result.stderr}')
            return result
        return result

    def rename(self, new_name: str) -> CommandResult:
        """Rename the device."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup rename {self.dm_name} {new_name}')
        if result.failed:
            logger.error(f'Failed to rename {self.dm_name} to {new_name}')
            return result

        # Update internal state
        old_name = self.dm_name
        self.dm_name = new_name
        self.path = Path(f'/dev/mapper/{new_name}')
        logger.debug(f'Successfully renamed device from {old_name} to {new_name}')
        return result

    def set_uuid(self, uuid: str) -> CommandResult:
        """Set the UUID of a device (immutable once set)."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup rename {self.dm_name} --setuuid {uuid}')
        if result.failed:
            logger.error(f'Failed to set UUID for {self.dm_name}: {result.stderr}')
            return result

        logger.debug(f'Successfully set UUID {uuid} for device {self.dm_name}')
        return result

    def reload_table(self, new_table: str) -> CommandResult:
        """Suspend, load a new table, and resume."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        # Suspend device
        self.suspend().assert_ok()

        # Load new table
        result = self._run(f'dmsetup load {self.dm_name} "{new_table}"')
        if result.failed:
            logger.error(f'Failed to load new table: {result.stderr}')
            self.resume()  # best-effort resume
            return result

        # Resume with new table
        self.resume().assert_ok()

        # Refresh from system
        self.refresh()
        logger.debug(f'Successfully reloaded table for {self.dm_name}')
        return result

    def load(
        self,
        table: str | None = None,
        table_file: str | None = None,
    ) -> CommandResult:
        """Load table into the inactive slot (use ``resume()`` to make it live).

        Exactly one of ``table`` or ``table_file`` must be provided.
        """
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        if table and table_file:
            raise ValueError('Cannot specify both table and table_file')

        if table:
            result = self._run(f'dmsetup load {self.dm_name} --table "{table}"')
        elif table_file:
            result = self._run(f'dmsetup load {self.dm_name} {table_file}')
        else:
            raise ValueError('Either table or table_file must be provided')

        if result.failed:
            logger.error(f'Failed to load table for {self.dm_name}: {result.stderr}')
            return result
        return result

    def get_table(
        self,
        *,
        concise: bool = False,
        target_type: str | None = None,
        showkeys: bool = False,
    ) -> str | None:
        """Get the current device table."""
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        cmd = f'dmsetup table {self.dm_name}'
        if concise:
            cmd += ' --concise'
        if target_type:
            cmd += f' --target {target_type}'
        if showkeys:
            cmd += ' --showkeys'

        result = run(cmd)
        if result.failed:
            logger.error(f'Failed to get table for {self.dm_name}')
            return None

        return result.stdout.strip()

    def wipe_table(
        self,
        *,
        force: bool = False,
        noflush: bool = False,
        nolockfs: bool = False,
    ) -> CommandResult:
        """Replace the table with one that fails all new I/O.

        Flushes in-flight I/O first, then replaces the table to release
        any devices held open.
        """
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        cmd = f'dmsetup wipe_table {self.dm_name}'
        if force:
            cmd += ' --force'
        if noflush:
            cmd += ' --noflush'
        if nolockfs:
            cmd += ' --nolockfs'

        result = self._run(cmd)
        if result.failed:
            logger.error(f'Failed to wipe table for {self.dm_name}: {result.stderr}')
            return result
        return result

    def mknodes(self) -> CommandResult:
        """Ensure ``/dev/mapper`` node for this device is correct."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup mknodes {self.dm_name}')
        if result.failed:
            logger.error(f'Failed to mknodes for {self.dm_name}')
            return result
        return result

    def setgeometry(
        self,
        cylinders: int,
        heads: int,
        sectors: int,
        start: int,
    ) -> CommandResult:
        """Set the device geometry (C/H/S)."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup setgeometry {self.dm_name} {cylinders} {heads} {sectors} {start}')
        if result.failed:
            logger.error(f'Failed to set geometry for {self.dm_name}: {result.stderr}')
            return result
        return result

    def wait(
        self,
        event_nr: int | None = None,
        *,
        noflush: bool = False,
    ) -> int | None:
        """Wait until the event counter exceeds ``event_nr``."""
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        cmd = f'dmsetup wait {self.dm_name}'
        if noflush:
            cmd += ' --noflush'
        if event_nr is not None:
            cmd += f' {event_nr}'

        result = run(cmd)
        if result.failed:
            logger.error(f'Failed to wait for {self.dm_name}')
            return None

        # Parse event number from output
        try:
            return int(result.stdout.strip())
        except ValueError:
            return None

    def measure(self) -> str | None:
        """Show IMA measurement data (debug only, does not trigger a measurement)."""
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        result = run(f'dmsetup measure {self.dm_name}')
        if result.failed:
            logger.error(f'Failed to measure {self.dm_name}')
            return None

        return result.stdout.strip()

    def mangle(self) -> CommandResult:
        """Ensure name/UUID contain only udev-safe characters, renaming if needed."""
        if not self.is_created or not self.dm_name:
            raise DmError('Device not created or dm_name not available')

        result = self._run(f'dmsetup mangle {self.dm_name}')
        if result.failed:
            logger.error(f'Failed to mangle {self.dm_name}')
            return result
        return result

    def splitname(self, subsystem: str = 'LVM') -> dict[str, str] | None:
        """Split device name into subsystem constituents (e.g. LVM VG/LV/Layer)."""
        if not self.is_created or not self.dm_name:
            logger.error('Device not created or dm_name not available')
            return None

        result = run(f'dmsetup splitname {self.dm_name} {subsystem}')
        if result.failed:
            logger.error(f'Failed to split name {self.dm_name}')
            return None

        # Parse output into dictionary
        output = result.stdout.strip()
        name_dict: dict[str, str] = {}
        for line in output.splitlines():
            if ':' in line:
                key, value = line.split(':', 1)
                name_dict[key.strip()] = value.strip()

        return name_dict

    @staticmethod
    def version() -> dict[str, str] | None:
        """Get dmsetup and driver version information."""
        result = run('dmsetup version')
        if result.failed:
            logger.error('Failed to get dmsetup version')
            return None

        output = result.stdout.strip()
        version_dict: dict[str, str] = {}
        for line in output.splitlines():
            if ':' in line:
                key, value = line.split(':', 1)
                version_dict[key.strip().lower().replace(' ', '_')] = value.strip()

        return version_dict

    @staticmethod
    def targets() -> list[dict[str, str]]:
        """Get names and versions of currently-loaded DM targets."""
        result = run('dmsetup targets')
        if result.failed:
            logger.error('Failed to get targets')
            return []

        targets_list: list[dict[str, str]] = []
        for line in result.stdout.strip().splitlines():
            parts = line.strip().split()
            if len(parts) >= 2:
                targets_list.append(
                    {
                        'name': parts[0],
                        'version': parts[1],
                    }
                )

        return targets_list

    @staticmethod
    def udevcreatecookie() -> str | None:
        """Create a udev synchronization cookie (must be released via ``udevreleasecookie()``)."""
        result = run('dmsetup udevcreatecookie')
        if result.failed:
            logger.error('Failed to create udev cookie')
            return None

        return result.stdout.strip()

    @staticmethod
    def udevreleasecookie(cookie: str | None = None) -> CommandResult:
        """Wait for pending udev processing and release the cookie.

        Args:
            cookie: Cookie to release (default: ``DM_UDEV_COOKIE`` env var)
        """
        cmd = 'dmsetup udevreleasecookie'
        if cookie:
            cmd += f' {cookie}'

        result = run(cmd)
        if result.failed:
            logger.error('Failed to release udev cookie')
            return result
        return result

    @staticmethod
    def udevcomplete(cookie: str) -> CommandResult:
        """Signal udev processing completion for the given cookie."""
        result = run(f'dmsetup udevcomplete {cookie}')
        if result.failed:
            logger.error(f'Failed to complete udev cookie {cookie}')
            return result
        return result

    @staticmethod
    def udevcomplete_all(age_in_minutes: int | None = None) -> CommandResult:
        """Remove all cookies older than ``age_in_minutes`` (resumes waiting processes)."""
        cmd = 'dmsetup udevcomplete_all'
        if age_in_minutes is not None:
            cmd += f' {age_in_minutes}'

        result = run(cmd)
        if result.failed:
            logger.error('Failed to complete all udev cookies')
            return result
        return result

    @staticmethod
    def udevcookie() -> list[str]:
        """List all existing udev cookies."""
        result = run('dmsetup udevcookie')
        if result.failed:
            logger.error('Failed to list udev cookies')
            return []

        return result.stdout.strip().splitlines()

    @staticmethod
    def udevflags(cookie: str) -> dict[str, str]:
        """Parse udev control flags encoded in a cookie."""
        result = run(f'dmsetup udevflags {cookie}')
        if result.failed:
            logger.error(f'Failed to get udev flags for cookie {cookie}')
            return {}

        flags: dict[str, str] = {}
        for line in result.stdout.strip().splitlines():
            if '=' in line:
                key, value = line.split('=', 1)
                flags[key.strip()] = value.strip().strip('\'"')

        return flags

    @classmethod
    def remove_all(cls, *, force: bool = False, deferred: bool = False) -> CommandResult:
        """Remove all device mapper devices (resets the driver). Use with care."""
        cmd = 'dmsetup remove_all'
        if force:
            cmd += ' --force'
        if deferred:
            cmd += ' --deferred'

        result = run(cmd)
        if result.failed:
            logger.error('Failed to remove all devices')
            return result
        return result

    @classmethod
    def mknodes_all(cls) -> CommandResult:
        """Sync all ``/dev/mapper`` nodes with loaded device-mapper devices."""
        result = run('dmsetup mknodes')
        if result.failed:
            logger.error('Failed to run mknodes')
            return result
        return result

    @classmethod
    def ls(
        cls,
        *,
        target_type: str | None = None,
        output_format: str = 'devno',
        tree: bool = False,
        tree_options: str | None = None,
        exec_cmd: str | None = None,
    ) -> list[str] | str:
        """List device mapper devices.

        Returns tree string if tree=True, otherwise list of device names.
        """
        cmd = 'dmsetup ls'
        if target_type:
            cmd += f' --target {target_type}'
        if output_format != 'devno':
            cmd += f' -o {output_format}'
        if tree:
            cmd += ' --tree'
            if tree_options:
                cmd += f' -o {tree_options}'
        if exec_cmd:
            cmd += f' --exec "{exec_cmd}"'

        result = run(cmd)
        if result.failed:
            logger.warning('No Device Mapper devices found')
            return [] if not tree else ''

        output = result.stdout.strip()

        if tree or exec_cmd:
            return output

        # Parse device names from output
        devices: list[str] = []
        for line in output.splitlines():
            if line.strip():
                parts = line.strip().split()
                if parts:
                    devices.append(parts[0])

        return devices

    @classmethod
    def get_all(cls) -> Sequence[DmDevice]:
        """Discover and return all Device Mapper devices on the system."""
        devices: list[DmDevice] = []
        result = run('dmsetup ls')
        if result.failed:
            logger.warning('No Device Mapper devices found')
            return []

        for line in result.stdout.splitlines():
            try:
                stripped_line = line.strip()
                if not stripped_line:
                    continue

                parts = stripped_line.split()
                if len(parts) < 2:
                    continue

                dm_name = parts[0]
                dev_id = parts[1].strip('()')
                major, minor = dev_id.split(':')

                # Resolve (major, minor) to the kernel device name (dm-N) by following the
                # /sys/dev/block/{major}:{minor} symlink, whose target's basename is the
                # kernel name. Avoids spawning a subprocess per device (`ls | grep`), which
                # was also fragile when the /dev/dm-* glob matched nothing.
                sys_path = cls.SYS_BLOCK_PATH / f'{major}:{minor}'
                if not sys_path.exists():
                    continue
                name = sys_path.resolve().name

                devices.append(cls(dm_name=dm_name, name=name, path=f'/dev/{name}').discover())
            except (ValueError, DeviceError):
                logger.exception('Failed to parse device info')
                continue

        return devices

    @classmethod
    def create_concise(cls, concise_spec: str) -> CommandResult:
        """Create devices from concise specification.

        Format: ``<name>,<uuid>,<minor>,<flags>,<table>[,<table>+][;<next device>...]``
        Separate multiple devices with semicolons. Escape commas/semicolons with backslash.
        """
        result = run(f'dmsetup create --concise "{concise_spec}"')
        if result.failed:
            logger.error(f'Failed to create devices from concise spec: {result.stderr}')
            return result
        return result

    @staticmethod
    def help(*, columns: bool = False) -> str | None:
        """Get dmsetup command help.

        Args:
            columns: Include report field list in the output
        """
        cmd = 'dmsetup help'
        if columns:
            cmd += ' --columns'

        result = run(cmd)
        # Note: dmsetup help returns exit code 0 but writes to stderr
        output = result.stdout.strip() or result.stderr.strip()
        return output or None

    @classmethod
    def get_by_name(cls, dm_name: str) -> DmDevice | None:
        """Look up a Device Mapper device by name, or return None."""
        if not dm_name:
            raise ValueError('Device Mapper name required')

        # Check if device exists
        result = run(f'dmsetup info {dm_name}')
        if result.failed:
            return None

        try:
            return cls(dm_name=dm_name).discover()
        except DeviceError:
            return None

    @staticmethod
    def _get_device_identifier(device: BlockDevice) -> str:
        """Get major:minor device identifier, calling ``discover()`` if needed."""
        device_id = device.device_id
        if device_id:
            return device_id
        # BlockDevice may not have been discover()-ed yet
        device.discover()
        device_id = device.device_id
        if device_id:
            return device_id
        raise DeviceError(f'No device ID available for {device.path}')

device_path property

Path to device in sysfs (/sys/class/block/<name>).

device_root_dir property

Device mapper root directory (/dev/mapper).

dm_device_path property

Full device path (e.g. /dev/mapper/my-device), or None if dm_name not set.

type property

Target type string (e.g. 'linear', 'delay', 'vdo').

__str__()

Return target table entry.

Format: Used in dmsetup table commands.

Source code in sts_libs/src/sts/dm/base.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_sectors} {self.type} {self.args}'

clear()

Destroy the table in the inactive table slot.

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

def clear(self) -> CommandResult:
    """Destroy the table in the inactive table slot."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup clear {self.dm_name}')
    if result.failed:
        logger.error(f'Failed to clear inactive table for {self.dm_name}')
        return result
    return result

create(dm_name, *, uuid=None, readonly=False, notable=False, readahead=None, addnodeoncreate=False, addnodeonresume=False)

Create the device on the system.

After successful creation, this device becomes a full BlockDevice with all properties populated from the system.

Parameters:

Name	Type	Description	Default
dm_name	str	Device mapper name	required
uuid	str | None	UUID for the device (can be used in place of name in commands)	None
readonly	bool	Load the table as read-only	False
notable	bool	Create device without loading any table	False
readahead	str | int | None	Sectors, or 'auto', 'none', or '+N' for minimum	None
addnodeoncreate	bool	Ensure /dev/mapper node exists after create	False
addnodeonresume	bool	Ensure /dev/mapper node exists after resume	False

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

def create(
    self,
    dm_name: str,
    *,
    uuid: str | None = None,
    readonly: bool = False,
    notable: bool = False,
    readahead: str | int | None = None,
    addnodeoncreate: bool = False,
    addnodeonresume: bool = False,
) -> CommandResult:
    """Create the device on the system.

    After successful creation, this device becomes a full BlockDevice
    with all properties populated from the system.

    Args:
        dm_name: Device mapper name
        uuid: UUID for the device (can be used in place of name in commands)
        readonly: Load the table as read-only
        notable: Create device without loading any table
        readahead: Sectors, or 'auto', 'none', or '+N' for minimum
        addnodeoncreate: Ensure /dev/mapper node exists after create
        addnodeonresume: Ensure /dev/mapper node exists after resume
    """
    if self.is_created:
        logger.warning(f'Device {self.dm_name} already created')
        return CommandResult(command=f'dmsetup create {dm_name}', rc=0, stdout='Already created')

    cmd = f'dmsetup create {dm_name}'

    if uuid:
        cmd += f' --uuid {uuid}'
    if readonly:
        cmd += ' --readonly'
    if addnodeoncreate:
        cmd += ' --addnodeoncreate'
    if addnodeonresume:
        cmd += ' --addnodeonresume'
    if readahead is not None:
        cmd += f' --readahead {readahead}'

    if notable:
        cmd += ' --notable'
    else:
        # Build table from this device's configuration
        table = str(self)
        cmd += f' --table "{table}"'
        logger.info(f'Creating device {dm_name} with table: {table}')

    result = self._run(cmd)

    if result.failed:
        logger.error(f'Failed to create device {dm_name}: {result.stderr}')
        return result

    logger.debug(f'Successfully created device mapper device: {dm_name}')

    # Set dm_name and reinitialize as existing device
    self.dm_name = dm_name
    self.path = Path(f'/dev/mapper/{dm_name}')
    self.name = None

    # Get kernel device name
    info_result = run(f'dmsetup info -c --noheadings -o name,blkdevname {dm_name}')
    if info_result.succeeded:
        parts = info_result.stdout.strip().split()
        if len(parts) >= 2:
            self.name = parts[1]

    # Load device state from system (BlockDevice properties, table)
    self.discover()
    return result

create_concise(concise_spec) classmethod

Create devices from concise specification.

Format: <name>,<uuid>,<minor>,<flags>,<table>[,<table>+][;<next device>...] Separate multiple devices with semicolons. Escape commas/semicolons with backslash.

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

@classmethod
def create_concise(cls, concise_spec: str) -> CommandResult:
    """Create devices from concise specification.

    Format: ``<name>,<uuid>,<minor>,<flags>,<table>[,<table>+][;<next device>...]``
    Separate multiple devices with semicolons. Escape commas/semicolons with backslash.
    """
    result = run(f'dmsetup create --concise "{concise_spec}"')
    if result.failed:
        logger.error(f'Failed to create devices from concise spec: {result.stderr}')
        return result
    return result

deps(*, output_format='devno')

Get devices referenced by the live table.

Parameters:

Name	Type	Description	Default
output_format	str	'devno' (major:minor), 'blkdevname', or 'devname'	'devno'

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

def deps(self, *, output_format: str = 'devno') -> list[str]:
    """Get devices referenced by the live table.

    Args:
        output_format: 'devno' (major:minor), 'blkdevname', or 'devname'
    """
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return []

    result = run(f'dmsetup deps -o {output_format} {self.dm_name}')
    if result.failed:
        logger.error(f'Failed to get dependencies for {self.dm_name}')
        return []

    # Parse output: "1 dependencies  : (major, minor) ..."
    output = result.stdout.strip()
    deps = []
    if ':' in output:
        deps_part = output.split(':', 1)[1].strip()
        # Extract device identifiers from the output
        if output_format == 'devno':
            # Format: (major, minor) or (major:minor)
            matches = re.findall(r'\((\d+[,:]\s*\d+)\)', deps_part)
            deps = [m.replace(' ', '').replace(',', ':') for m in matches]
        else:
            # Format: (device_name)
            matches = re.findall(r'\(([^)]+)\)', deps_part)
            deps = matches

    return deps

discover()

Load device state from the system.

Call after construction for devices that already exist (identified by dm_name, name, or path). No-op for configuration-only devices.

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

def discover(self) -> Self:
    """Load device state from the system.

    Call after construction for devices that already exist (identified by
    dm_name, name, or path). No-op for configuration-only devices.
    """
    if not self.path and not self.dm_name and not self.name:
        return self

    super().discover()

    # 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()

    # Load table from system
    if self.dm_name:
        result = run(f'dmsetup table {self.dm_name}')
        if result.succeeded:
            self.table = result.stdout.strip()
            self._parse_table()

    self.is_created = True
    return self

get_all() classmethod

Discover and return all Device Mapper devices on the system.

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

@classmethod
def get_all(cls) -> Sequence[DmDevice]:
    """Discover and return all Device Mapper devices on the system."""
    devices: list[DmDevice] = []
    result = run('dmsetup ls')
    if result.failed:
        logger.warning('No Device Mapper devices found')
        return []

    for line in result.stdout.splitlines():
        try:
            stripped_line = line.strip()
            if not stripped_line:
                continue

            parts = stripped_line.split()
            if len(parts) < 2:
                continue

            dm_name = parts[0]
            dev_id = parts[1].strip('()')
            major, minor = dev_id.split(':')

            # Resolve (major, minor) to the kernel device name (dm-N) by following the
            # /sys/dev/block/{major}:{minor} symlink, whose target's basename is the
            # kernel name. Avoids spawning a subprocess per device (`ls | grep`), which
            # was also fragile when the /dev/dm-* glob matched nothing.
            sys_path = cls.SYS_BLOCK_PATH / f'{major}:{minor}'
            if not sys_path.exists():
                continue
            name = sys_path.resolve().name

            devices.append(cls(dm_name=dm_name, name=name, path=f'/dev/{name}').discover())
        except (ValueError, DeviceError):
            logger.exception('Failed to parse device info')
            continue

    return devices

get_by_name(dm_name) classmethod

Look up a Device Mapper device by name, or return None.

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

@classmethod
def get_by_name(cls, dm_name: str) -> DmDevice | None:
    """Look up a Device Mapper device by name, or return None."""
    if not dm_name:
        raise ValueError('Device Mapper name required')

    # Check if device exists
    result = run(f'dmsetup info {dm_name}')
    if result.failed:
        return None

    try:
        return cls(dm_name=dm_name).discover()
    except DeviceError:
        return None

get_status(*, target_type=None, noflush=False)

Get device status from dmsetup.

Parameters:

Name	Type	Description	Default
target_type	str | None	Only display information for the specified target type	None
noflush	bool	Do not commit thin-pool metadata before reporting	False

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

def get_status(
    self,
    *,
    target_type: str | None = None,
    noflush: bool = False,
) -> str | None:
    """Get device status from dmsetup.

    Args:
        target_type: Only display information for the specified target type
        noflush: Do not commit thin-pool metadata before reporting
    """
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    cmd = f'dmsetup status {self.dm_name}'
    if target_type:
        cmd += f' --target {target_type}'
    if noflush:
        cmd += ' --noflush'

    result = run(cmd)
    if result.failed:
        logger.error(f'Failed to get status for {self.dm_name}')
        return None

    return result.stdout.strip()

get_table(*, concise=False, target_type=None, showkeys=False)

Get the current device table.

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

def get_table(
    self,
    *,
    concise: bool = False,
    target_type: str | None = None,
    showkeys: bool = False,
) -> str | None:
    """Get the current device table."""
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    cmd = f'dmsetup table {self.dm_name}'
    if concise:
        cmd += ' --concise'
    if target_type:
        cmd += f' --target {target_type}'
    if showkeys:
        cmd += ' --showkeys'

    result = run(cmd)
    if result.failed:
        logger.error(f'Failed to get table for {self.dm_name}')
        return None

    return result.stdout.strip()

help(*, columns=False) staticmethod

Get dmsetup command help.

Parameters:

Name	Type	Description	Default
columns	bool	Include report field list in the output	False

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

@staticmethod
def help(*, columns: bool = False) -> str | None:
    """Get dmsetup command help.

    Args:
        columns: Include report field list in the output
    """
    cmd = 'dmsetup help'
    if columns:
        cmd += ' --columns'

    result = run(cmd)
    # Note: dmsetup help returns exit code 0 but writes to stderr
    output = result.stdout.strip() or result.stderr.strip()
    return output or None

info(*, columns=False, noheadings=False, fields=None, separator=None, sort_fields=None, nameprefixes=False)

Get device information.

Returns dict of field->value when columns=False, raw string when columns=True, or None on error.

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

def info(
    self,
    *,
    columns: bool = False,
    noheadings: bool = False,
    fields: str | None = None,
    separator: str | None = None,
    sort_fields: str | None = None,
    nameprefixes: bool = False,
) -> dict[str, str] | str | None:
    """Get device information.

    Returns dict of field->value when columns=False, raw string when
    columns=True, or None on error.
    """
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    cmd = f'dmsetup info {self.dm_name}'
    if columns:
        cmd += ' --columns'
        if noheadings:
            cmd += ' --noheadings'
        if fields:
            cmd += f' -o {fields}'
        if separator:
            cmd += f' --separator "{separator}"'
        if sort_fields:
            cmd += f' --sort {sort_fields}'
        if nameprefixes:
            cmd += ' --nameprefixes'

    result = run(cmd)
    if result.failed:
        logger.error(f'Failed to get info for {self.dm_name}')
        return None

    output = result.stdout.strip()

    if columns:
        return output

    # Parse Field: Value format into dictionary
    info_dict: dict[str, str] = {}
    for line in output.splitlines():
        if ':' in line:
            key, value = line.split(':', 1)
            info_dict[key.strip()] = value.strip()

    return info_dict

load(table=None, table_file=None)

Load table into the inactive slot (use resume() to make it live).

Exactly one of table or table_file must be provided.

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

def load(
    self,
    table: str | None = None,
    table_file: str | None = None,
) -> CommandResult:
    """Load table into the inactive slot (use ``resume()`` to make it live).

    Exactly one of ``table`` or ``table_file`` must be provided.
    """
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    if table and table_file:
        raise ValueError('Cannot specify both table and table_file')

    if table:
        result = self._run(f'dmsetup load {self.dm_name} --table "{table}"')
    elif table_file:
        result = self._run(f'dmsetup load {self.dm_name} {table_file}')
    else:
        raise ValueError('Either table or table_file must be provided')

    if result.failed:
        logger.error(f'Failed to load table for {self.dm_name}: {result.stderr}')
        return result
    return result

ls(*, target_type=None, output_format='devno', tree=False, tree_options=None, exec_cmd=None) classmethod

List device mapper devices.

Returns tree string if tree=True, otherwise list of device names.

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

@classmethod
def ls(
    cls,
    *,
    target_type: str | None = None,
    output_format: str = 'devno',
    tree: bool = False,
    tree_options: str | None = None,
    exec_cmd: str | None = None,
) -> list[str] | str:
    """List device mapper devices.

    Returns tree string if tree=True, otherwise list of device names.
    """
    cmd = 'dmsetup ls'
    if target_type:
        cmd += f' --target {target_type}'
    if output_format != 'devno':
        cmd += f' -o {output_format}'
    if tree:
        cmd += ' --tree'
        if tree_options:
            cmd += f' -o {tree_options}'
    if exec_cmd:
        cmd += f' --exec "{exec_cmd}"'

    result = run(cmd)
    if result.failed:
        logger.warning('No Device Mapper devices found')
        return [] if not tree else ''

    output = result.stdout.strip()

    if tree or exec_cmd:
        return output

    # Parse device names from output
    devices: list[str] = []
    for line in output.splitlines():
        if line.strip():
            parts = line.strip().split()
            if parts:
                devices.append(parts[0])

    return devices

mangle()

Ensure name/UUID contain only udev-safe characters, renaming if needed.

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

def mangle(self) -> CommandResult:
    """Ensure name/UUID contain only udev-safe characters, renaming if needed."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup mangle {self.dm_name}')
    if result.failed:
        logger.error(f'Failed to mangle {self.dm_name}')
        return result
    return result

measure()

Show IMA measurement data (debug only, does not trigger a measurement).

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

def measure(self) -> str | None:
    """Show IMA measurement data (debug only, does not trigger a measurement)."""
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    result = run(f'dmsetup measure {self.dm_name}')
    if result.failed:
        logger.error(f'Failed to measure {self.dm_name}')
        return None

    return result.stdout.strip()

message(sector, message)

Send a message to the target at the given sector (use 0 if irrelevant).

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

def message(self, sector: int, message: str) -> CommandResult:
    """Send a message to the target at the given sector (use 0 if irrelevant)."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup message {self.dm_name} {sector} {message}')
    if result.failed:
        logger.error(f'Failed to send message to {self.dm_name}: {result.stderr}')
        return result
    return result

mknodes()

Ensure /dev/mapper node for this device is correct.

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

def mknodes(self) -> CommandResult:
    """Ensure ``/dev/mapper`` node for this device is correct."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup mknodes {self.dm_name}')
    if result.failed:
        logger.error(f'Failed to mknodes for {self.dm_name}')
        return result
    return result

mknodes_all() classmethod

Sync all /dev/mapper nodes with loaded device-mapper devices.

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

@classmethod
def mknodes_all(cls) -> CommandResult:
    """Sync all ``/dev/mapper`` nodes with loaded device-mapper devices."""
    result = run('dmsetup mknodes')
    if result.failed:
        logger.error('Failed to run mknodes')
        return result
    return result

refresh()

Refresh device table from the system.

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

def refresh(self) -> None:
    """Refresh device table from the system."""
    if not self.is_created or not self.dm_name:
        return

    # Refresh table from system
    result = run(f'dmsetup table {self.dm_name}')
    if result.succeeded:
        self.table = result.stdout.strip()
        self._parse_table()

reload_table(new_table)

Suspend, load a new table, and resume.

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

def reload_table(self, new_table: str) -> CommandResult:
    """Suspend, load a new table, and resume."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    # Suspend device
    self.suspend().assert_ok()

    # Load new table
    result = self._run(f'dmsetup load {self.dm_name} "{new_table}"')
    if result.failed:
        logger.error(f'Failed to load new table: {result.stderr}')
        self.resume()  # best-effort resume
        return result

    # Resume with new table
    self.resume().assert_ok()

    # Refresh from system
    self.refresh()
    logger.debug(f'Successfully reloaded table for {self.dm_name}')
    return result

remove(*, force=False, retry=False, deferred=False)

Remove the device mapping.

Parameters:

Name	Type	Description	Default
force	bool	Replace the table with one that fails all I/O	False
retry	bool	Retry removal for a few seconds if it fails (e.g., udev race)	False
deferred	bool	Device removed when last user closes it	False

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

def remove(
    self,
    *,
    force: bool = False,
    retry: bool = False,
    deferred: bool = False,
) -> CommandResult:
    """Remove the device mapping.

    Args:
        force: Replace the table with one that fails all I/O
        retry: Retry removal for a few seconds if it fails (e.g., udev race)
        deferred: Device removed when last user closes it
    """
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    cmd = f'dmsetup remove {self.dm_name}'
    if force:
        cmd += ' --force'
    if retry:
        cmd += ' --retry'
    if deferred:
        cmd += ' --deferred'

    result = self._run(cmd)
    if result.failed:
        logger.error('Failed to remove device')
        return result

    self.is_created = False
    return result

remove_all(*, force=False, deferred=False) classmethod

Remove all device mapper devices (resets the driver). Use with care.

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

@classmethod
def remove_all(cls, *, force: bool = False, deferred: bool = False) -> CommandResult:
    """Remove all device mapper devices (resets the driver). Use with care."""
    cmd = 'dmsetup remove_all'
    if force:
        cmd += ' --force'
    if deferred:
        cmd += ' --deferred'

    result = run(cmd)
    if result.failed:
        logger.error('Failed to remove all devices')
        return result
    return result

rename(new_name)

Rename the device.

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

def rename(self, new_name: str) -> CommandResult:
    """Rename the device."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup rename {self.dm_name} {new_name}')
    if result.failed:
        logger.error(f'Failed to rename {self.dm_name} to {new_name}')
        return result

    # Update internal state
    old_name = self.dm_name
    self.dm_name = new_name
    self.path = Path(f'/dev/mapper/{new_name}')
    logger.debug(f'Successfully renamed device from {old_name} to {new_name}')
    return result

resume(*, addnodeoncreate=False, addnodeonresume=False, noflush=False, nolockfs=False, readahead=None)

Resume a suspended device.

If an inactive table has been loaded, it becomes live.

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

def resume(
    self,
    *,
    addnodeoncreate: bool = False,
    addnodeonresume: bool = False,
    noflush: bool = False,
    nolockfs: bool = False,
    readahead: str | int | None = None,
) -> CommandResult:
    """Resume a suspended device.

    If an inactive table has been loaded, it becomes live.
    """
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    cmd = f'dmsetup resume {self.dm_name}'
    if addnodeoncreate:
        cmd += ' --addnodeoncreate'
    if addnodeonresume:
        cmd += ' --addnodeonresume'
    if noflush:
        cmd += ' --noflush'
    if nolockfs:
        cmd += ' --nolockfs'
    if readahead is not None:
        cmd += f' --readahead {readahead}'

    result = self._run(cmd)
    if result.failed:
        logger.error('Failed to resume device')
        return result
    return result

set_uuid(uuid)

Set the UUID of a device (immutable once set).

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

def set_uuid(self, uuid: str) -> CommandResult:
    """Set the UUID of a device (immutable once set)."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup rename {self.dm_name} --setuuid {uuid}')
    if result.failed:
        logger.error(f'Failed to set UUID for {self.dm_name}: {result.stderr}')
        return result

    logger.debug(f'Successfully set UUID {uuid} for device {self.dm_name}')
    return result

setgeometry(cylinders, heads, sectors, start)

Set the device geometry (C/H/S).

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

def setgeometry(
    self,
    cylinders: int,
    heads: int,
    sectors: int,
    start: int,
) -> CommandResult:
    """Set the device geometry (C/H/S)."""
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    result = self._run(f'dmsetup setgeometry {self.dm_name} {cylinders} {heads} {sectors} {start}')
    if result.failed:
        logger.error(f'Failed to set geometry for {self.dm_name}: {result.stderr}')
        return result
    return result

splitname(subsystem='LVM')

Split device name into subsystem constituents (e.g. LVM VG/LV/Layer).

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

def splitname(self, subsystem: str = 'LVM') -> dict[str, str] | None:
    """Split device name into subsystem constituents (e.g. LVM VG/LV/Layer)."""
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    result = run(f'dmsetup splitname {self.dm_name} {subsystem}')
    if result.failed:
        logger.error(f'Failed to split name {self.dm_name}')
        return None

    # Parse output into dictionary
    output = result.stdout.strip()
    name_dict: dict[str, str] = {}
    for line in output.splitlines():
        if ':' in line:
            key, value = line.split(':', 1)
            name_dict[key.strip()] = value.strip()

    return name_dict

suspend(*, nolockfs=False, noflush=False)

Suspend I/O to the device.

Pending I/O is flushed; new I/O is postponed until resume().

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

def suspend(
    self,
    *,
    nolockfs: bool = False,
    noflush: bool = False,
) -> CommandResult:
    """Suspend I/O to the device.

    Pending I/O is flushed; new I/O is postponed until ``resume()``.
    """
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    cmd = f'dmsetup suspend {self.dm_name}'
    if nolockfs:
        cmd += ' --nolockfs'
    if noflush:
        cmd += ' --noflush'

    result = self._run(cmd)
    if result.failed:
        logger.error('Failed to suspend device')
        return result
    return result

targets() staticmethod

Get names and versions of currently-loaded DM targets.

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

@staticmethod
def targets() -> list[dict[str, str]]:
    """Get names and versions of currently-loaded DM targets."""
    result = run('dmsetup targets')
    if result.failed:
        logger.error('Failed to get targets')
        return []

    targets_list: list[dict[str, str]] = []
    for line in result.stdout.strip().splitlines():
        parts = line.strip().split()
        if len(parts) >= 2:
            targets_list.append(
                {
                    'name': parts[0],
                    'version': parts[1],
                }
            )

    return targets_list

udevcomplete(cookie) staticmethod

Signal udev processing completion for the given cookie.

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

@staticmethod
def udevcomplete(cookie: str) -> CommandResult:
    """Signal udev processing completion for the given cookie."""
    result = run(f'dmsetup udevcomplete {cookie}')
    if result.failed:
        logger.error(f'Failed to complete udev cookie {cookie}')
        return result
    return result

udevcomplete_all(age_in_minutes=None) staticmethod

Remove all cookies older than age_in_minutes (resumes waiting processes).

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

@staticmethod
def udevcomplete_all(age_in_minutes: int | None = None) -> CommandResult:
    """Remove all cookies older than ``age_in_minutes`` (resumes waiting processes)."""
    cmd = 'dmsetup udevcomplete_all'
    if age_in_minutes is not None:
        cmd += f' {age_in_minutes}'

    result = run(cmd)
    if result.failed:
        logger.error('Failed to complete all udev cookies')
        return result
    return result

udevcookie() staticmethod

List all existing udev cookies.

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

@staticmethod
def udevcookie() -> list[str]:
    """List all existing udev cookies."""
    result = run('dmsetup udevcookie')
    if result.failed:
        logger.error('Failed to list udev cookies')
        return []

    return result.stdout.strip().splitlines()

udevcreatecookie() staticmethod

Create a udev synchronization cookie (must be released via udevreleasecookie()).

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

@staticmethod
def udevcreatecookie() -> str | None:
    """Create a udev synchronization cookie (must be released via ``udevreleasecookie()``)."""
    result = run('dmsetup udevcreatecookie')
    if result.failed:
        logger.error('Failed to create udev cookie')
        return None

    return result.stdout.strip()

udevflags(cookie) staticmethod

Parse udev control flags encoded in a cookie.

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

@staticmethod
def udevflags(cookie: str) -> dict[str, str]:
    """Parse udev control flags encoded in a cookie."""
    result = run(f'dmsetup udevflags {cookie}')
    if result.failed:
        logger.error(f'Failed to get udev flags for cookie {cookie}')
        return {}

    flags: dict[str, str] = {}
    for line in result.stdout.strip().splitlines():
        if '=' in line:
            key, value = line.split('=', 1)
            flags[key.strip()] = value.strip().strip('\'"')

    return flags

udevreleasecookie(cookie=None) staticmethod

Wait for pending udev processing and release the cookie.

Parameters:

Name	Type	Description	Default
cookie	str | None	Cookie to release (default: DM_UDEV_COOKIE env var)	None

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

@staticmethod
def udevreleasecookie(cookie: str | None = None) -> CommandResult:
    """Wait for pending udev processing and release the cookie.

    Args:
        cookie: Cookie to release (default: ``DM_UDEV_COOKIE`` env var)
    """
    cmd = 'dmsetup udevreleasecookie'
    if cookie:
        cmd += f' {cookie}'

    result = run(cmd)
    if result.failed:
        logger.error('Failed to release udev cookie')
        return result
    return result

version() staticmethod

Get dmsetup and driver version information.

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

@staticmethod
def version() -> dict[str, str] | None:
    """Get dmsetup and driver version information."""
    result = run('dmsetup version')
    if result.failed:
        logger.error('Failed to get dmsetup version')
        return None

    output = result.stdout.strip()
    version_dict: dict[str, str] = {}
    for line in output.splitlines():
        if ':' in line:
            key, value = line.split(':', 1)
            version_dict[key.strip().lower().replace(' ', '_')] = value.strip()

    return version_dict

wait(event_nr=None, *, noflush=False)

Wait until the event counter exceeds event_nr.

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

def wait(
    self,
    event_nr: int | None = None,
    *,
    noflush: bool = False,
) -> int | None:
    """Wait until the event counter exceeds ``event_nr``."""
    if not self.is_created or not self.dm_name:
        logger.error('Device not created or dm_name not available')
        return None

    cmd = f'dmsetup wait {self.dm_name}'
    if noflush:
        cmd += ' --noflush'
    if event_nr is not None:
        cmd += f' {event_nr}'

    result = run(cmd)
    if result.failed:
        logger.error(f'Failed to wait for {self.dm_name}')
        return None

    # Parse event number from output
    try:
        return int(result.stdout.strip())
    except ValueError:
        return None

wipe_table(*, force=False, noflush=False, nolockfs=False)

Replace the table with one that fails all new I/O.

Flushes in-flight I/O first, then replaces the table to release any devices held open.

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

def wipe_table(
    self,
    *,
    force: bool = False,
    noflush: bool = False,
    nolockfs: bool = False,
) -> CommandResult:
    """Replace the table with one that fails all new I/O.

    Flushes in-flight I/O first, then replaces the table to release
    any devices held open.
    """
    if not self.is_created or not self.dm_name:
        raise DmError('Device not created or dm_name not available')

    cmd = f'dmsetup wipe_table {self.dm_name}'
    if force:
        cmd += ' --force'
    if noflush:
        cmd += ' --noflush'
    if nolockfs:
        cmd += ' --nolockfs'

    result = self._run(cmd)
    if result.failed:
        logger.error(f'Failed to wipe table for {self.dm_name}: {result.stderr}')
        return result
    return result

sts.dm.cache

Device Mapper cache target.

dm-cache improves performance of a block device (e.g., a spindle) by dynamically migrating some of its data to a faster, smaller device (e.g., an SSD).

The target requires three devices: 1. Origin device - the big, slow one (e.g., HDD) 2. Cache device - the small, fast one (e.g., SSD) 3. Metadata device - records which blocks are in cache

Table format

cache <#feature args> [] <#policy args> [policy args]

Cache modes

• writeback (default): writes go only to cache, marked dirty
• writethrough: writes go to both cache and origin
• passthrough: all I/O goes to origin, useful for coherency

Policies

• default: alias for best performing policy
• smq: stochastic multi-queue (recommended)
• mq: multi-queue (legacy)

CacheDevice pydantic-model

Bases: DmDevice

Cache target -- caches data from a slow origin on a fast device.

Args format: <metadata> <cache> <origin> <block_size> <#features> [features] <policy> <#policy_args> [args]

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Cache target -- caches data from a slow origin on a fast device.\n\nArgs format: ``<metadata> <cache> <origin> <block_size> <#features> [features] <policy> <#policy_args> [args]``",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "cache",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "CacheDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• target_type (str)

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

class CacheDevice(DmDevice):
    """Cache target -- caches data from a slow origin on a fast device.

    Args format: ``<metadata> <cache> <origin> <block_size> <#features> [features] <policy> <#policy_args> [args]``
    """

    target_type: str = Field(default='cache', init=False)

    @classmethod
    def from_block_devices(
        cls,
        metadata_device: BlockDevice,
        cache_device: BlockDevice,
        origin_device: BlockDevice,
        block_size_sectors: int = 512,
        policy: str = 'default',
        policy_args: dict[str, str] | None = None,
        start: int = 0,
        size_sectors: int | None = None,
        *,
        writethrough: bool = False,
        passthrough: bool = False,
        metadata2: bool = False,
        no_discard_passdown: bool = False,
    ) -> CacheDevice:
        """Create CacheDevice from BlockDevices.

        Args:
            metadata_device: Device for cache metadata
            cache_device: Fast device (e.g. SSD)
            origin_device: Slow device (e.g. HDD)
            block_size_sectors: Block size in sectors (default: 512 = 256KB).
                Must be between 64 (32KB) and 2097152 (1GB), multiple of 64.
            policy: Cache policy name
            policy_args: Policy-specific key/value arguments
            start: Start sector in virtual device
            size_sectors: Size in sectors (default: origin device size)
            writethrough: Writes go to both cache and origin
            passthrough: All I/O to origin
            metadata2: Use version 2 metadata format
            no_discard_passdown: Don't pass discards to origin
        """
        metadata_id = cls._get_device_identifier(metadata_device)
        cache_id = cls._get_device_identifier(cache_device)
        origin_id = cls._get_device_identifier(origin_device)

        if size_sectors is None and origin_device.size is not None:
            size_sectors = origin_device.size // origin_device.sector_size

        # Build feature list
        features: list[str] = []
        if writethrough:
            features.append('writethrough')
        if passthrough:
            features.append('passthrough')
        if metadata2:
            features.append('metadata2')
        if no_discard_passdown:
            features.append('no_discard_passdown')

        # Build policy args list
        policy_args_list: list[str] = []
        if policy_args:
            for key, value in policy_args.items():
                policy_args_list.extend([key, value])

        # Build args: <metadata> <cache> <origin> <block_size> <#features> [features] <policy> <#policy_args> [args]
        args_parts = [
            metadata_id,
            cache_id,
            origin_id,
            str(block_size_sectors),
            str(len(features)),
        ]
        args_parts.extend(features)
        args_parts.extend((policy, str(len(policy_args_list))))
        args_parts.extend(policy_args_list)

        args = ' '.join(args_parts)
        if size_sectors is None:
            raise ValueError('size_sectors must be provided or origin_device.size must be available')
        return cls(start=start, size_sectors=size_sectors, args=args)

    @staticmethod
    def parse_status(status: str) -> CacheStatus:
        """Parse cache device status string.

        Raw status format (from ``dmsetup status``)::

            <start> <size> cache <metadata_block_size> <used_meta>/<total_meta>
            <cache_block_size> <used_cache>/<total_cache> <read_hits> <read_misses>
            <write_hits> <write_misses> <demotions> <promotions> <dirty>
            <#features> [features] <#core_args> [core_args] <policy>
            <#policy_args> [policy_args] <mode> <needs_check>
        """
        parts = status.split()
        result: dict[str, str | int] = {}

        # Status includes table header: <start> <size> cache <actual status...>
        # Skip first 3 fields to get to actual cache status
        if len(parts) >= 14 and parts[2] == 'cache':
            offset = 3  # Skip: start, size, "cache"
            result['metadata_block_size'] = int(parts[offset])
            if '/' in parts[offset + 1]:
                used, total = parts[offset + 1].split('/')
                result['used_metadata_blocks'] = int(used)
                result['total_metadata_blocks'] = int(total)
            result['cache_block_size'] = int(parts[offset + 2])
            if '/' in parts[offset + 3]:
                used, total = parts[offset + 3].split('/')
                result['used_cache_blocks'] = int(used)
                result['total_cache_blocks'] = int(total)
            result['read_hits'] = int(parts[offset + 4])
            result['read_misses'] = int(parts[offset + 5])
            result['write_hits'] = int(parts[offset + 6])
            result['write_misses'] = int(parts[offset + 7])
            result['demotions'] = int(parts[offset + 8])
            result['promotions'] = int(parts[offset + 9])
            result['dirty'] = int(parts[offset + 10])

        return CacheStatus.model_validate(result)

from_block_devices(metadata_device, cache_device, origin_device, block_size_sectors=512, policy='default', policy_args=None, start=0, size_sectors=None, *, writethrough=False, passthrough=False, metadata2=False, no_discard_passdown=False) classmethod

Create CacheDevice from BlockDevices.

Parameters:

Name	Type	Description	Default
metadata_device	BlockDevice	Device for cache metadata	required
cache_device	BlockDevice	Fast device (e.g. SSD)	required
origin_device	BlockDevice	Slow device (e.g. HDD)	required
block_size_sectors	int	Block size in sectors (default: 512 = 256KB). Must be between 64 (32KB) and 2097152 (1GB), multiple of 64.	512
policy	str	Cache policy name	'default'
policy_args	dict[str, str] | None	Policy-specific key/value arguments	None
start	int	Start sector in virtual device	0
size_sectors	int | None	Size in sectors (default: origin device size)	None
writethrough	bool	Writes go to both cache and origin	False
passthrough	bool	All I/O to origin	False
metadata2	bool	Use version 2 metadata format	False
no_discard_passdown	bool	Don't pass discards to origin	False

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

@classmethod
def from_block_devices(
    cls,
    metadata_device: BlockDevice,
    cache_device: BlockDevice,
    origin_device: BlockDevice,
    block_size_sectors: int = 512,
    policy: str = 'default',
    policy_args: dict[str, str] | None = None,
    start: int = 0,
    size_sectors: int | None = None,
    *,
    writethrough: bool = False,
    passthrough: bool = False,
    metadata2: bool = False,
    no_discard_passdown: bool = False,
) -> CacheDevice:
    """Create CacheDevice from BlockDevices.

    Args:
        metadata_device: Device for cache metadata
        cache_device: Fast device (e.g. SSD)
        origin_device: Slow device (e.g. HDD)
        block_size_sectors: Block size in sectors (default: 512 = 256KB).
            Must be between 64 (32KB) and 2097152 (1GB), multiple of 64.
        policy: Cache policy name
        policy_args: Policy-specific key/value arguments
        start: Start sector in virtual device
        size_sectors: Size in sectors (default: origin device size)
        writethrough: Writes go to both cache and origin
        passthrough: All I/O to origin
        metadata2: Use version 2 metadata format
        no_discard_passdown: Don't pass discards to origin
    """
    metadata_id = cls._get_device_identifier(metadata_device)
    cache_id = cls._get_device_identifier(cache_device)
    origin_id = cls._get_device_identifier(origin_device)

    if size_sectors is None and origin_device.size is not None:
        size_sectors = origin_device.size // origin_device.sector_size

    # Build feature list
    features: list[str] = []
    if writethrough:
        features.append('writethrough')
    if passthrough:
        features.append('passthrough')
    if metadata2:
        features.append('metadata2')
    if no_discard_passdown:
        features.append('no_discard_passdown')

    # Build policy args list
    policy_args_list: list[str] = []
    if policy_args:
        for key, value in policy_args.items():
            policy_args_list.extend([key, value])

    # Build args: <metadata> <cache> <origin> <block_size> <#features> [features] <policy> <#policy_args> [args]
    args_parts = [
        metadata_id,
        cache_id,
        origin_id,
        str(block_size_sectors),
        str(len(features)),
    ]
    args_parts.extend(features)
    args_parts.extend((policy, str(len(policy_args_list))))
    args_parts.extend(policy_args_list)

    args = ' '.join(args_parts)
    if size_sectors is None:
        raise ValueError('size_sectors must be provided or origin_device.size must be available')
    return cls(start=start, size_sectors=size_sectors, args=args)

parse_status(status) staticmethod

Parse cache device status string.

Raw status format (from dmsetup status)::

<start> <size> cache <metadata_block_size> <used_meta>/<total_meta>
<cache_block_size> <used_cache>/<total_cache> <read_hits> <read_misses>
<write_hits> <write_misses> <demotions> <promotions> <dirty>
<#features> [features] <#core_args> [core_args] <policy>
<#policy_args> [policy_args] <mode> <needs_check>

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

@staticmethod
def parse_status(status: str) -> CacheStatus:
    """Parse cache device status string.

    Raw status format (from ``dmsetup status``)::

        <start> <size> cache <metadata_block_size> <used_meta>/<total_meta>
        <cache_block_size> <used_cache>/<total_cache> <read_hits> <read_misses>
        <write_hits> <write_misses> <demotions> <promotions> <dirty>
        <#features> [features] <#core_args> [core_args] <policy>
        <#policy_args> [policy_args] <mode> <needs_check>
    """
    parts = status.split()
    result: dict[str, str | int] = {}

    # Status includes table header: <start> <size> cache <actual status...>
    # Skip first 3 fields to get to actual cache status
    if len(parts) >= 14 and parts[2] == 'cache':
        offset = 3  # Skip: start, size, "cache"
        result['metadata_block_size'] = int(parts[offset])
        if '/' in parts[offset + 1]:
            used, total = parts[offset + 1].split('/')
            result['used_metadata_blocks'] = int(used)
            result['total_metadata_blocks'] = int(total)
        result['cache_block_size'] = int(parts[offset + 2])
        if '/' in parts[offset + 3]:
            used, total = parts[offset + 3].split('/')
            result['used_cache_blocks'] = int(used)
            result['total_cache_blocks'] = int(total)
        result['read_hits'] = int(parts[offset + 4])
        result['read_misses'] = int(parts[offset + 5])
        result['write_hits'] = int(parts[offset + 6])
        result['write_misses'] = int(parts[offset + 7])
        result['demotions'] = int(parts[offset + 8])
        result['promotions'] = int(parts[offset + 9])
        result['dirty'] = int(parts[offset + 10])

    return CacheStatus.model_validate(result)

CacheStatus pydantic-model

Bases: ReportModel

Parsed 'dmsetup status' output for a cache target.

See CacheDevice.parse_status for the raw status line format.

Show JSON schema:

{
  "description": "Parsed 'dmsetup status' output for a cache target.\n\nSee `CacheDevice.parse_status` for the raw status line format.",
  "properties": {
    "metadata_block_size": {
      "default": 0,
      "title": "Metadata Block Size",
      "type": "integer"
    },
    "used_metadata_blocks": {
      "default": 0,
      "title": "Used Metadata Blocks",
      "type": "integer"
    },
    "total_metadata_blocks": {
      "default": 0,
      "title": "Total Metadata Blocks",
      "type": "integer"
    },
    "cache_block_size": {
      "default": 0,
      "title": "Cache Block Size",
      "type": "integer"
    },
    "used_cache_blocks": {
      "default": 0,
      "title": "Used Cache Blocks",
      "type": "integer"
    },
    "total_cache_blocks": {
      "default": 0,
      "title": "Total Cache Blocks",
      "type": "integer"
    },
    "read_hits": {
      "default": 0,
      "title": "Read Hits",
      "type": "integer"
    },
    "read_misses": {
      "default": 0,
      "title": "Read Misses",
      "type": "integer"
    },
    "write_hits": {
      "default": 0,
      "title": "Write Hits",
      "type": "integer"
    },
    "write_misses": {
      "default": 0,
      "title": "Write Misses",
      "type": "integer"
    },
    "demotions": {
      "default": 0,
      "title": "Demotions",
      "type": "integer"
    },
    "promotions": {
      "default": 0,
      "title": "Promotions",
      "type": "integer"
    },
    "dirty": {
      "default": 0,
      "title": "Dirty",
      "type": "integer"
    }
  },
  "title": "CacheStatus",
  "type": "object"
}

Fields:

• metadata_block_size (int)
• used_metadata_blocks (int)
• total_metadata_blocks (int)
• cache_block_size (int)
• used_cache_blocks (int)
• total_cache_blocks (int)
• read_hits (int)
• read_misses (int)
• write_hits (int)
• write_misses (int)
• demotions (int)
• promotions (int)
• dirty (int)

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

class CacheStatus(ReportModel):
    """Parsed 'dmsetup status' output for a cache target.

    See `CacheDevice.parse_status` for the raw status line format.
    """

    metadata_block_size: int = 0
    used_metadata_blocks: int = 0
    total_metadata_blocks: int = 0
    cache_block_size: int = 0
    used_cache_blocks: int = 0
    total_cache_blocks: int = 0
    read_hits: int = 0
    read_misses: int = 0
    write_hits: int = 0
    write_misses: int = 0
    demotions: int = 0
    promotions: int = 0
    dirty: int = 0

sts.dm.delay

Device Mapper delay target.

DelayDevice pydantic-model

Bases: DmDevice

Delay target -- delays reads/writes/flushes, optionally to different devices.

Table line has either 3, 6, or 9 arguments:

• 3 args: <device> <offset> <delay> -- same delay for all operations
• 6 args: ... <write_device> <write_offset> <write_delay> -- separate write/flush delay
• 9 args: ... <flush_device> <flush_offset> <flush_delay> -- separate flush delay

Offsets are in sectors, delays in milliseconds. Parsed fields (read_device, write_device, flush_device, etc.) are populated by _parse_table() and arg_format indicates which variant ('3', '6', or '9') was used.

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Delay target -- delays reads/writes/flushes, optionally to different devices.\n\nTable line has either 3, 6, or 9 arguments:\n\n- 3 args: ``<device> <offset> <delay>`` -- same delay for all operations\n- 6 args: ``... <write_device> <write_offset> <write_delay>`` -- separate write/flush delay\n- 9 args: ``... <flush_device> <flush_offset> <flush_delay>`` -- separate flush delay\n\nOffsets are in sectors, delays in milliseconds. Parsed fields (read_device,\nwrite_device, flush_device, etc.) are populated by ``_parse_table()`` and\n``arg_format`` indicates which variant ('3', '6', or '9') was used.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "delay",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    },
    "read_device": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Read Device"
    },
    "read_offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Read Offset"
    },
    "read_delay_ms": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Read Delay Ms"
    },
    "write_device": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Write Device"
    },
    "write_offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Write Offset"
    },
    "write_delay_ms": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Write Delay Ms"
    },
    "flush_device": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Flush Device"
    },
    "flush_offset": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Flush Offset"
    },
    "flush_delay_ms": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Flush Delay Ms"
    },
    "arg_format": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Arg Format"
    }
  },
  "title": "DelayDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• read_device (str | None)
• read_offset (int | None)
• read_delay_ms (int | None)
• write_device (str | None)
• write_offset (int | None)
• write_delay_ms (int | None)
• flush_device (str | None)
• flush_offset (int | None)
• flush_delay_ms (int | None)
• arg_format (str | None)
• target_type (str)

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

class DelayDevice(DmDevice):
    """Delay target -- delays reads/writes/flushes, optionally to different devices.

    Table line has either 3, 6, or 9 arguments:

    - 3 args: ``<device> <offset> <delay>`` -- same delay for all operations
    - 6 args: ``... <write_device> <write_offset> <write_delay>`` -- separate write/flush delay
    - 9 args: ``... <flush_device> <flush_offset> <flush_delay>`` -- separate flush delay

    Offsets are in sectors, delays in milliseconds. Parsed fields (read_device,
    write_device, flush_device, etc.) are populated by ``_parse_table()`` and
    ``arg_format`` indicates which variant ('3', '6', or '9') was used.
    """

    # Parsed attributes (populated by refresh)
    read_device: str | None = Field(default=None, init=False)
    read_offset: int | None = Field(default=None, init=False)
    read_delay_ms: int | None = Field(default=None, init=False)
    write_device: str | None = Field(default=None, init=False)
    write_offset: int | None = Field(default=None, init=False)
    write_delay_ms: int | None = Field(default=None, init=False)
    flush_device: str | None = Field(default=None, init=False)
    flush_offset: int | None = Field(default=None, init=False)
    flush_delay_ms: int | None = Field(default=None, init=False)
    arg_format: str | None = Field(default=None, init=False)

    target_type: str = Field(default='delay', init=False)

    @classmethod
    def from_block_device(
        cls,
        device: BlockDevice,
        delay_ms: int,
        start: int = 0,
        size_sectors: int | None = None,
        offset: int = 0,
    ) -> DelayDevice:
        """Create DelayDevice from BlockDevice (3 argument format).

        Applies the same delay to read, write, and flush operations.
        """
        device_id = cls._get_device_identifier(device)

        if size_sectors is None and device.size is not None:
            size_sectors = device.size // device.sector_size

        args = f'{device_id} {offset} {delay_ms}'

        if size_sectors is None:
            raise ValueError('size_sectors must be provided or device.size must be available')
        return cls(start=start, size_sectors=size_sectors, args=args)

    @classmethod
    def from_block_devices_rw(
        cls,
        read_device: BlockDevice,
        read_offset: int,
        read_delay_ms: int,
        write_device: BlockDevice,
        write_offset: int,
        write_delay_ms: int,
        start: int = 0,
        size: int | None = None,
    ) -> DelayDevice:
        """Create DelayDevice with separate read and write/flush devices (6 argument format)."""
        read_device_id = cls._get_device_identifier(read_device)
        write_device_id = cls._get_device_identifier(write_device)

        if size is None and read_device.size is not None:
            size = read_device.size // read_device.sector_size

        args = f'{read_device_id} {read_offset} {read_delay_ms} {write_device_id} {write_offset} {write_delay_ms}'

        if size is None:
            raise ValueError('size must be provided or read_device.size must be available')
        return cls(start=start, size_sectors=size, args=args)

    @classmethod
    def from_block_devices_rwf(
        cls,
        read_device: BlockDevice,
        read_offset: int,
        read_delay_ms: int,
        write_device: BlockDevice,
        write_offset: int,
        write_delay_ms: int,
        flush_device: BlockDevice,
        flush_offset: int,
        flush_delay_ms: int,
        start: int = 0,
        size: int | None = None,
    ) -> DelayDevice:
        """Create DelayDevice with separate read, write, and flush devices (9 argument format)."""
        read_device_id = cls._get_device_identifier(read_device)
        write_device_id = cls._get_device_identifier(write_device)
        flush_device_id = cls._get_device_identifier(flush_device)

        if size is None and read_device.size is not None:
            size = read_device.size // read_device.sector_size

        args = (
            f'{read_device_id} {read_offset} {read_delay_ms} '
            f'{write_device_id} {write_offset} {write_delay_ms} '
            f'{flush_device_id} {flush_offset} {flush_delay_ms}'
        )

        if size is None:
            raise ValueError('size must be provided or read_device.size must be available')
        return cls(start=start, size_sectors=size, args=args)

    @classmethod
    def create_positional(
        cls,
        device_path: str,
        offset: int,
        delay_ms: int,
        start: int = 0,
        size: int | None = None,
        write_device_path: str | None = None,
        write_offset: int | None = None,
        write_delay_ms: int | None = None,
        flush_device_path: str | None = None,
        flush_offset: int | None = None,
        flush_delay_ms: int | None = None,
    ) -> DelayDevice:
        """Create DelayDevice with positional arguments.

        Supports 3, 6, or 9 argument formats depending on which
        write/flush parameters are provided.
        """
        if size is None:
            raise ValueError('Size must be specified for delay targets')

        # Build args based on provided parameters
        args = f'{device_path} {offset} {delay_ms}'

        # Check for 6-argument format
        if write_device_path is not None:
            if write_offset is None or write_delay_ms is None:
                raise ValueError('write_offset and write_delay_ms required when write_device_path is specified')
            args = f'{args} {write_device_path} {write_offset} {write_delay_ms}'

            # Check for 9-argument format
            if flush_device_path is not None:
                if flush_offset is None or flush_delay_ms is None:
                    raise ValueError('flush_offset and flush_delay_ms required when flush_device_path is specified')
                args = f'{args} {flush_device_path} {flush_offset} {flush_delay_ms}'
        elif flush_device_path is not None:
            raise ValueError('flush_device_path requires write_device_path to be specified first')

        return cls(start=start, size_sectors=size, args=args)

    def _parse_table(self) -> None:
        """Parse delay-specific attributes from the args string."""
        super()._parse_table()
        if not self.args:
            return

        parts = self.args.split()

        if len(parts) < 3:
            return

        # Parse read device params (always present)
        self.read_device = parts[0]
        self.read_offset = int(parts[1])
        self.read_delay_ms = int(parts[2])

        if len(parts) == 3:
            self.arg_format = '3'
        elif len(parts) >= 6:
            # Parse write device params
            self.write_device = parts[3]
            self.write_offset = int(parts[4])
            self.write_delay_ms = int(parts[5])

            if len(parts) == 6:
                self.arg_format = '6'
            elif len(parts) >= 9:
                # Parse flush device params
                self.flush_device = parts[6]
                self.flush_offset = int(parts[7])
                self.flush_delay_ms = int(parts[8])
                self.arg_format = '9'

    @classmethod
    def from_table_line(cls, table_line: str) -> DelayDevice | None:
        """Create DelayDevice from a ``dmsetup table`` output line."""
        parts = table_line.strip().split(None, 3)
        if len(parts) < 4:
            logger.warning(f'Invalid table line: {table_line}')
            return None

        start = int(parts[0])
        size = int(parts[1])
        target_type = parts[2]
        args = parts[3]

        if target_type != 'delay':
            logger.warning(f'Not a delay target: {target_type}')
            return None

        target = cls(start=start, size_sectors=size, args=args)
        target._parse_table()  # parse args to populate attributes
        return target

create_positional(device_path, offset, delay_ms, start=0, size=None, write_device_path=None, write_offset=None, write_delay_ms=None, flush_device_path=None, flush_offset=None, flush_delay_ms=None) classmethod

Create DelayDevice with positional arguments.

Supports 3, 6, or 9 argument formats depending on which write/flush parameters are provided.

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

@classmethod
def create_positional(
    cls,
    device_path: str,
    offset: int,
    delay_ms: int,
    start: int = 0,
    size: int | None = None,
    write_device_path: str | None = None,
    write_offset: int | None = None,
    write_delay_ms: int | None = None,
    flush_device_path: str | None = None,
    flush_offset: int | None = None,
    flush_delay_ms: int | None = None,
) -> DelayDevice:
    """Create DelayDevice with positional arguments.

    Supports 3, 6, or 9 argument formats depending on which
    write/flush parameters are provided.
    """
    if size is None:
        raise ValueError('Size must be specified for delay targets')

    # Build args based on provided parameters
    args = f'{device_path} {offset} {delay_ms}'

    # Check for 6-argument format
    if write_device_path is not None:
        if write_offset is None or write_delay_ms is None:
            raise ValueError('write_offset and write_delay_ms required when write_device_path is specified')
        args = f'{args} {write_device_path} {write_offset} {write_delay_ms}'

        # Check for 9-argument format
        if flush_device_path is not None:
            if flush_offset is None or flush_delay_ms is None:
                raise ValueError('flush_offset and flush_delay_ms required when flush_device_path is specified')
            args = f'{args} {flush_device_path} {flush_offset} {flush_delay_ms}'
    elif flush_device_path is not None:
        raise ValueError('flush_device_path requires write_device_path to be specified first')

    return cls(start=start, size_sectors=size, args=args)

from_block_device(device, delay_ms, start=0, size_sectors=None, offset=0) classmethod

Create DelayDevice from BlockDevice (3 argument format).

Applies the same delay to read, write, and flush operations.

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

@classmethod
def from_block_device(
    cls,
    device: BlockDevice,
    delay_ms: int,
    start: int = 0,
    size_sectors: int | None = None,
    offset: int = 0,
) -> DelayDevice:
    """Create DelayDevice from BlockDevice (3 argument format).

    Applies the same delay to read, write, and flush operations.
    """
    device_id = cls._get_device_identifier(device)

    if size_sectors is None and device.size is not None:
        size_sectors = device.size // device.sector_size

    args = f'{device_id} {offset} {delay_ms}'

    if size_sectors is None:
        raise ValueError('size_sectors must be provided or device.size must be available')
    return cls(start=start, size_sectors=size_sectors, args=args)

from_block_devices_rw(read_device, read_offset, read_delay_ms, write_device, write_offset, write_delay_ms, start=0, size=None) classmethod

Create DelayDevice with separate read and write/flush devices (6 argument format).

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

@classmethod
def from_block_devices_rw(
    cls,
    read_device: BlockDevice,
    read_offset: int,
    read_delay_ms: int,
    write_device: BlockDevice,
    write_offset: int,
    write_delay_ms: int,
    start: int = 0,
    size: int | None = None,
) -> DelayDevice:
    """Create DelayDevice with separate read and write/flush devices (6 argument format)."""
    read_device_id = cls._get_device_identifier(read_device)
    write_device_id = cls._get_device_identifier(write_device)

    if size is None and read_device.size is not None:
        size = read_device.size // read_device.sector_size

    args = f'{read_device_id} {read_offset} {read_delay_ms} {write_device_id} {write_offset} {write_delay_ms}'

    if size is None:
        raise ValueError('size must be provided or read_device.size must be available')
    return cls(start=start, size_sectors=size, args=args)

from_block_devices_rwf(read_device, read_offset, read_delay_ms, write_device, write_offset, write_delay_ms, flush_device, flush_offset, flush_delay_ms, start=0, size=None) classmethod

Create DelayDevice with separate read, write, and flush devices (9 argument format).

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

@classmethod
def from_block_devices_rwf(
    cls,
    read_device: BlockDevice,
    read_offset: int,
    read_delay_ms: int,
    write_device: BlockDevice,
    write_offset: int,
    write_delay_ms: int,
    flush_device: BlockDevice,
    flush_offset: int,
    flush_delay_ms: int,
    start: int = 0,
    size: int | None = None,
) -> DelayDevice:
    """Create DelayDevice with separate read, write, and flush devices (9 argument format)."""
    read_device_id = cls._get_device_identifier(read_device)
    write_device_id = cls._get_device_identifier(write_device)
    flush_device_id = cls._get_device_identifier(flush_device)

    if size is None and read_device.size is not None:
        size = read_device.size // read_device.sector_size

    args = (
        f'{read_device_id} {read_offset} {read_delay_ms} '
        f'{write_device_id} {write_offset} {write_delay_ms} '
        f'{flush_device_id} {flush_offset} {flush_delay_ms}'
    )

    if size is None:
        raise ValueError('size must be provided or read_device.size must be available')
    return cls(start=start, size_sectors=size, args=args)

from_table_line(table_line) classmethod

Create DelayDevice from a dmsetup table output line.

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

@classmethod
def from_table_line(cls, table_line: str) -> DelayDevice | None:
    """Create DelayDevice from a ``dmsetup table`` output line."""
    parts = table_line.strip().split(None, 3)
    if len(parts) < 4:
        logger.warning(f'Invalid table line: {table_line}')
        return None

    start = int(parts[0])
    size = int(parts[1])
    target_type = parts[2]
    args = parts[3]

    if target_type != 'delay':
        logger.warning(f'Not a delay target: {target_type}')
        return None

    target = cls(start=start, size_sectors=size, args=args)
    target._parse_table()  # parse args to populate attributes
    return target

sts.dm.error

Device Mapper error target.

ErrorDevice pydantic-model

Bases: DmDevice

Error target -- returns I/O errors for all operations.

Takes no target arguments.

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Error target -- returns I/O errors for all operations.\n\nTakes no target arguments.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "error",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "ErrorDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• target_type (str)

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

class ErrorDevice(DmDevice):
    """Error target -- returns I/O errors for all operations.

    Takes no target arguments.
    """

    target_type: str = Field(default='error', init=False)

    @classmethod
    def create_config(
        cls,
        start: int = 0,
        size: int | None = None,
    ) -> ErrorDevice:
        """Create ErrorDevice configuration."""
        if size is None:
            raise ValueError('Size must be specified for error targets')

        return cls(start=start, size_sectors=size, args='')

create_config(start=0, size=None) classmethod

Create ErrorDevice configuration.

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

@classmethod
def create_config(
    cls,
    start: int = 0,
    size: int | None = None,
) -> ErrorDevice:
    """Create ErrorDevice configuration."""
    if size is None:
        raise ValueError('Size must be specified for error targets')

    return cls(start=start, size_sectors=size, args='')

sts.dm.flakey

Device Mapper flakey target.

The flakey target is similar to linear but exhibits unreliable behavior periodically. It's useful for simulating failing devices for testing.

Starting from when the table is loaded, the device is available for seconds, then exhibits unreliable behavior for seconds, and then this cycle repeats.

Table format

[ []]

Features

• drop_writes: Silently ignore all writes, reads work normally
• error_writes: Fail all writes with error, reads work normally
• corrupt_bio_byte : Corrupt specific bytes

FlakeyDevice pydantic-model

Bases: DmDevice

Flakey target -- simulates an unreliable device that periodically fails I/O.

Args format: <device> <offset> <up_interval> <down_interval> [<num_features> [<features>]]

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Flakey target -- simulates an unreliable device that periodically fails I/O.\n\nArgs format: ``<device> <offset> <up_interval> <down_interval> [<num_features> [<features>]]``",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "flakey",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "FlakeyDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• target_type (str)

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

class FlakeyDevice(DmDevice):
    """Flakey target -- simulates an unreliable device that periodically fails I/O.

    Args format: ``<device> <offset> <up_interval> <down_interval> [<num_features> [<features>]]``
    """

    target_type: str = Field(default='flakey', init=False)

    @classmethod
    def from_block_device(
        cls,
        device: BlockDevice,
        up_interval: int,
        down_interval: int,
        offset: int = 0,
        size_sectors: int | None = None,
        corrupt_bio_byte: tuple[int, str, int, int] | None = None,
        start: int = 0,
        *,
        drop_writes: bool = False,
        error_writes: bool = False,
    ) -> FlakeyDevice:
        """Create FlakeyDevice from BlockDevice.

        Args:
            device: Underlying block device
            up_interval: Seconds device is available (reliable)
            down_interval: Seconds device is unreliable
            offset: Starting sector in underlying device
            size_sectors: Size in sectors (default: full device)
            corrupt_bio_byte: Tuple of (nth_byte, direction, value, flags)
            start: Start sector in virtual device
            drop_writes: Silently drop writes when down
            error_writes: Fail writes with error when down
        """
        device_id = cls._get_device_identifier(device)

        if size_sectors is None and device.size is not None:
            size_sectors = device.size // device.sector_size

        # Build args: <device> <offset> <up_interval> <down_interval> [features]
        args_parts = [
            device_id,
            str(offset),
            str(up_interval),
            str(down_interval),
        ]

        # Build feature list
        features: list[str] = []
        if drop_writes:
            features.append('drop_writes')
        if error_writes:
            features.append('error_writes')
        if corrupt_bio_byte is not None:
            nth_byte, direction, value, flags = corrupt_bio_byte
            features.append(f'corrupt_bio_byte {nth_byte} {direction} {value} {flags}')

        # Add features if any
        if features:
            args_parts.append(str(len(features)))
            args_parts.extend(features)

        args = ' '.join(args_parts)
        if size_sectors is None:
            raise ValueError('size_sectors must be provided or device.size must be available')
        return cls(start=start, size_sectors=size_sectors, args=args)

from_block_device(device, up_interval, down_interval, offset=0, size_sectors=None, corrupt_bio_byte=None, start=0, *, drop_writes=False, error_writes=False) classmethod

Create FlakeyDevice from BlockDevice.

Parameters:

Name	Type	Description	Default
device	BlockDevice	Underlying block device	required
up_interval	int	Seconds device is available (reliable)	required
down_interval	int	Seconds device is unreliable	required
offset	int	Starting sector in underlying device	0
size_sectors	int | None	Size in sectors (default: full device)	None
corrupt_bio_byte	tuple[int, str, int, int] | None	Tuple of (nth_byte, direction, value, flags)	None
start	int	Start sector in virtual device	0
drop_writes	bool	Silently drop writes when down	False
error_writes	bool	Fail writes with error when down	False

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

@classmethod
def from_block_device(
    cls,
    device: BlockDevice,
    up_interval: int,
    down_interval: int,
    offset: int = 0,
    size_sectors: int | None = None,
    corrupt_bio_byte: tuple[int, str, int, int] | None = None,
    start: int = 0,
    *,
    drop_writes: bool = False,
    error_writes: bool = False,
) -> FlakeyDevice:
    """Create FlakeyDevice from BlockDevice.

    Args:
        device: Underlying block device
        up_interval: Seconds device is available (reliable)
        down_interval: Seconds device is unreliable
        offset: Starting sector in underlying device
        size_sectors: Size in sectors (default: full device)
        corrupt_bio_byte: Tuple of (nth_byte, direction, value, flags)
        start: Start sector in virtual device
        drop_writes: Silently drop writes when down
        error_writes: Fail writes with error when down
    """
    device_id = cls._get_device_identifier(device)

    if size_sectors is None and device.size is not None:
        size_sectors = device.size // device.sector_size

    # Build args: <device> <offset> <up_interval> <down_interval> [features]
    args_parts = [
        device_id,
        str(offset),
        str(up_interval),
        str(down_interval),
    ]

    # Build feature list
    features: list[str] = []
    if drop_writes:
        features.append('drop_writes')
    if error_writes:
        features.append('error_writes')
    if corrupt_bio_byte is not None:
        nth_byte, direction, value, flags = corrupt_bio_byte
        features.append(f'corrupt_bio_byte {nth_byte} {direction} {value} {flags}')

    # Add features if any
    if features:
        args_parts.append(str(len(features)))
        args_parts.extend(features)

    args = ' '.join(args_parts)
    if size_sectors is None:
        raise ValueError('size_sectors must be provided or device.size must be available')
    return cls(start=start, size_sectors=size_sectors, args=args)

sts.dm.linear

Device Mapper linear target.

LinearDevice pydantic-model

Bases: DmDevice

Linear target -- maps a linear range onto another device.

Args format: <destination device> <sector offset>

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Linear target -- maps a linear range onto another device.\n\nArgs format: ``<destination device> <sector offset>``",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "linear",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "LinearDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• target_type (str)

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

class LinearDevice(DmDevice):
    """Linear target -- maps a linear range onto another device.

    Args format: ``<destination device> <sector offset>``
    """

    target_type: str = Field(default='linear', init=False)

    @classmethod
    def from_block_device(
        cls,
        device: BlockDevice,
        start: int = 0,
        size_sectors: int | None = None,
        offset: int = 0,
    ) -> LinearDevice:
        """Create LinearDevice from BlockDevice."""
        device_id = cls._get_device_identifier(device)

        if size_sectors is None and device.size is not None:
            size_sectors = device.size // device.sector_size

        args = f'{device_id} {offset}'

        if size_sectors is None:
            raise ValueError('size_sectors must be provided or device.size must be available')
        return cls(start=start, size_sectors=size_sectors, args=args)

    @classmethod
    def create_positional(
        cls,
        device_path: str,
        offset: int = 0,
        start: int = 0,
        size_sectors: int | None = None,
    ) -> LinearDevice:
        """Create LinearDevice with positional arguments."""
        if size_sectors is None:
            raise ValueError('Size must be specified for linear devices')

        args = f'{device_path} {offset}'
        return cls(start=start, size_sectors=size_sectors, args=args)

create_positional(device_path, offset=0, start=0, size_sectors=None) classmethod

Create LinearDevice with positional arguments.

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

@classmethod
def create_positional(
    cls,
    device_path: str,
    offset: int = 0,
    start: int = 0,
    size_sectors: int | None = None,
) -> LinearDevice:
    """Create LinearDevice with positional arguments."""
    if size_sectors is None:
        raise ValueError('Size must be specified for linear devices')

    args = f'{device_path} {offset}'
    return cls(start=start, size_sectors=size_sectors, args=args)

from_block_device(device, start=0, size_sectors=None, offset=0) classmethod

Create LinearDevice from BlockDevice.

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

@classmethod
def from_block_device(
    cls,
    device: BlockDevice,
    start: int = 0,
    size_sectors: int | None = None,
    offset: int = 0,
) -> LinearDevice:
    """Create LinearDevice from BlockDevice."""
    device_id = cls._get_device_identifier(device)

    if size_sectors is None and device.size is not None:
        size_sectors = device.size // device.sector_size

    args = f'{device_id} {offset}'

    if size_sectors is None:
        raise ValueError('size_sectors must be provided or device.size must be available')
    return cls(start=start, size_sectors=size_sectors, args=args)

sts.dm.thin

Device Mapper thin provisioning targets (thin-pool and thin).

ThinDevice pydantic-model

Bases: DmDevice

Thin target -- allocates space from a thin pool on demand.

Maintains a reference to its parent pool for lifecycle management.

Args format: <pool dev> <dev id>

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    },
    "ThinDevice": {
      "additionalProperties": false,
      "description": "Thin target -- allocates space from a thin pool on demand.\n\nMaintains a reference to its parent pool for lifecycle management.\n\nArgs format: ``<pool dev> <dev id>``",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "size": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "blockdev_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/BlockdevInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "lsblk_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/LsblkInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size_sectors": {
          "default": 0,
          "title": "Size Sectors",
          "type": "integer"
        },
        "args": {
          "default": "",
          "title": "Args",
          "type": "string"
        },
        "dm_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Dm Name"
        },
        "target_type": {
          "default": "thin",
          "title": "Target Type",
          "type": "string"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Table"
        },
        "is_created": {
          "default": false,
          "title": "Is Created",
          "type": "boolean"
        },
        "pool": {
          "anyOf": [
            {
              "$ref": "#/$defs/ThinPoolDevice"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "thin_id": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Thin Id"
        }
      },
      "title": "ThinDevice",
      "type": "object"
    },
    "ThinPoolDevice": {
      "additionalProperties": false,
      "description": "Thin pool target -- manages a pool from which thin volumes are allocated.\n\nTracks child ThinDevice instances and provides methods for creating,\ndeleting, and snapshotting thin volumes.\n\nArgs format: ``<metadata dev> <data dev> <block size> <low water mark> <flags> <args>``\n\nExample:\n    ```python\n    pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev)\n    pool.create('my-pool')\n    thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin')\n    snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap')\n    pool.delete_thin(1)\n    pool.delete_thin(0)\n    pool.remove()\n    ```",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "size": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "blockdev_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/BlockdevInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "lsblk_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/LsblkInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size_sectors": {
          "default": 0,
          "title": "Size Sectors",
          "type": "integer"
        },
        "args": {
          "default": "",
          "title": "Args",
          "type": "string"
        },
        "dm_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Dm Name"
        },
        "target_type": {
          "default": "thin-pool",
          "title": "Target Type",
          "type": "string"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Table"
        },
        "is_created": {
          "default": false,
          "title": "Is Created",
          "type": "boolean"
        },
        "thin_devices": {
          "additionalProperties": {
            "$ref": "#/$defs/ThinDevice"
          },
          "title": "Thin Devices",
          "type": "object"
        }
      },
      "title": "ThinPoolDevice",
      "type": "object"
    }
  },
  "$ref": "#/$defs/ThinDevice"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• pool (ThinPoolDevice | None)
• thin_id (int | None)
• target_type (str)

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

class ThinDevice(DmDevice):
    """Thin target -- allocates space from a thin pool on demand.

    Maintains a reference to its parent pool for lifecycle management.

    Args format: ``<pool dev> <dev id>``
    """

    # Reference to parent pool
    pool: ThinPoolDevice | None = Field(default=None, repr=False)

    # Thin device ID within pool
    thin_id: int | None = None

    target_type: str = Field(default='thin', init=False)

    @classmethod
    def create_from_pool(
        cls,
        pool: ThinPoolDevice,
        thin_id: int,
        size: int,
        start: int = 0,
    ) -> ThinDevice:
        """Create ThinDevice configuration from pool.

        Used by ``ThinPoolDevice.create_thin()`` and ``create_snapshot()``
        to build device configuration before activation.
        """
        pool_id = cls._get_device_identifier(pool)
        args = f'{pool_id} {thin_id}'

        return cls(start=start, size_sectors=size, args=args, pool=pool, thin_id=thin_id)

    @classmethod
    def from_thin_pool(
        cls,
        pool_device: DmDevice,
        thin_id: int,
        start: int = 0,
        size: int | None = None,
    ) -> ThinDevice:
        """Create ThinDevice from thin pool device.

        For full parent-child tracking, prefer ``ThinPoolDevice.create_thin()``.
        """
        if size is None:
            raise ValueError('Size must be specified for thin targets')

        pool_id = cls._get_device_identifier(pool_device)
        args = f'{pool_id} {thin_id}'

        # Set pool reference if it's a ThinPoolDevice
        pool_ref = pool_device if isinstance(pool_device, ThinPoolDevice) else None

        return cls(start=start, size_sectors=size, args=args, thin_id=thin_id, pool=pool_ref)

create_from_pool(pool, thin_id, size, start=0) classmethod

Create ThinDevice configuration from pool.

Used by ThinPoolDevice.create_thin() and create_snapshot() to build device configuration before activation.

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

@classmethod
def create_from_pool(
    cls,
    pool: ThinPoolDevice,
    thin_id: int,
    size: int,
    start: int = 0,
) -> ThinDevice:
    """Create ThinDevice configuration from pool.

    Used by ``ThinPoolDevice.create_thin()`` and ``create_snapshot()``
    to build device configuration before activation.
    """
    pool_id = cls._get_device_identifier(pool)
    args = f'{pool_id} {thin_id}'

    return cls(start=start, size_sectors=size, args=args, pool=pool, thin_id=thin_id)

from_thin_pool(pool_device, thin_id, start=0, size=None) classmethod

Create ThinDevice from thin pool device.

For full parent-child tracking, prefer ThinPoolDevice.create_thin().

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

@classmethod
def from_thin_pool(
    cls,
    pool_device: DmDevice,
    thin_id: int,
    start: int = 0,
    size: int | None = None,
) -> ThinDevice:
    """Create ThinDevice from thin pool device.

    For full parent-child tracking, prefer ``ThinPoolDevice.create_thin()``.
    """
    if size is None:
        raise ValueError('Size must be specified for thin targets')

    pool_id = cls._get_device_identifier(pool_device)
    args = f'{pool_id} {thin_id}'

    # Set pool reference if it's a ThinPoolDevice
    pool_ref = pool_device if isinstance(pool_device, ThinPoolDevice) else None

    return cls(start=start, size_sectors=size, args=args, thin_id=thin_id, pool=pool_ref)

ThinPoolDevice pydantic-model

Bases: DmDevice

Thin pool target -- manages a pool from which thin volumes are allocated.

Tracks child ThinDevice instances and provides methods for creating, deleting, and snapshotting thin volumes.

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

Example

pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev)
pool.create('my-pool')
thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin')
snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap')
pool.delete_thin(1)
pool.delete_thin(0)
pool.remove()

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    },
    "ThinDevice": {
      "additionalProperties": false,
      "description": "Thin target -- allocates space from a thin pool on demand.\n\nMaintains a reference to its parent pool for lifecycle management.\n\nArgs format: ``<pool dev> <dev id>``",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "size": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "blockdev_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/BlockdevInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "lsblk_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/LsblkInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size_sectors": {
          "default": 0,
          "title": "Size Sectors",
          "type": "integer"
        },
        "args": {
          "default": "",
          "title": "Args",
          "type": "string"
        },
        "dm_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Dm Name"
        },
        "target_type": {
          "default": "thin",
          "title": "Target Type",
          "type": "string"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Table"
        },
        "is_created": {
          "default": false,
          "title": "Is Created",
          "type": "boolean"
        },
        "pool": {
          "anyOf": [
            {
              "$ref": "#/$defs/ThinPoolDevice"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "thin_id": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Thin Id"
        }
      },
      "title": "ThinDevice",
      "type": "object"
    },
    "ThinPoolDevice": {
      "additionalProperties": false,
      "description": "Thin pool target -- manages a pool from which thin volumes are allocated.\n\nTracks child ThinDevice instances and provides methods for creating,\ndeleting, and snapshotting thin volumes.\n\nArgs format: ``<metadata dev> <data dev> <block size> <low water mark> <flags> <args>``\n\nExample:\n    ```python\n    pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev)\n    pool.create('my-pool')\n    thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin')\n    snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap')\n    pool.delete_thin(1)\n    pool.delete_thin(0)\n    pool.remove()\n    ```",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "size": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "blockdev_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/BlockdevInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "lsblk_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/LsblkInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size_sectors": {
          "default": 0,
          "title": "Size Sectors",
          "type": "integer"
        },
        "args": {
          "default": "",
          "title": "Args",
          "type": "string"
        },
        "dm_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Dm Name"
        },
        "target_type": {
          "default": "thin-pool",
          "title": "Target Type",
          "type": "string"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Table"
        },
        "is_created": {
          "default": false,
          "title": "Is Created",
          "type": "boolean"
        },
        "thin_devices": {
          "additionalProperties": {
            "$ref": "#/$defs/ThinDevice"
          },
          "title": "Thin Devices",
          "type": "object"
        }
      },
      "title": "ThinPoolDevice",
      "type": "object"
    }
  },
  "$ref": "#/$defs/ThinPoolDevice"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• thin_devices (dict[int, ThinDevice])
• target_type (str)

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

class ThinPoolDevice(DmDevice):
    """Thin pool target -- manages a pool from which thin volumes are allocated.

    Tracks child ThinDevice instances and provides methods for creating,
    deleting, and snapshotting thin volumes.

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

    Example:
        ```python
        pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev)
        pool.create('my-pool')
        thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin')
        snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap')
        pool.delete_thin(1)
        pool.delete_thin(0)
        pool.remove()
        ```
    """

    # Track child thin devices: thin_id -> ThinDevice
    thin_devices: dict[int, ThinDevice] = Field(default_factory=dict, init=False, repr=False)

    target_type: str = Field(default='thin-pool', init=False)

    @classmethod
    def from_block_devices(
        cls,
        metadata_device: BlockDevice,
        data_device: BlockDevice,
        block_size_sectors: int = 128,
        low_water_mark: int = 0,
        features: list[str] | None = None,
        start: int = 0,
        size: int | None = None,
    ) -> ThinPoolDevice:
        """Create ThinPoolDevice from BlockDevices."""
        if features is None:
            features = ['skip_block_zeroing']

        metadata_id = cls._get_device_identifier(metadata_device)
        data_id = cls._get_device_identifier(data_device)

        if size is None and data_device.size is not None:
            size = data_device.size // data_device.sector_size

        # Build args: <metadata dev> <data dev> <block size> <low water mark> <# feature args> <feature args>
        args_parts = [
            metadata_id,
            data_id,
            str(block_size_sectors),
            str(low_water_mark),
            str(len(features)),
            *features,
        ]

        args = ' '.join(args_parts)
        if size is None:
            raise ValueError('size must be provided or data_device.size must be available')
        return cls(start=start, size_sectors=size, args=args)

    def create_thin(
        self,
        thin_id: int,
        size: int,
        dm_name: str,
        *,
        activate: bool = True,
    ) -> ThinDevice | None:
        """Create a new thin volume in the pool.

        Sends 'create_thin' message to the pool and optionally activates the device.
        """
        if not self.is_created:
            logger.error('Pool must be created before creating thin devices')
            return None

        if thin_id in self.thin_devices:
            logger.error(f'Thin device with ID {thin_id} already exists in pool')
            return None

        # Send create_thin message to pool
        if self.message(0, f'"create_thin {thin_id}"').failed:
            logger.error(f'Failed to create thin device {thin_id} in pool')
            return None

        logger.debug(f'Created thin device ID {thin_id} in pool {self.dm_name}')

        # Create ThinDevice configuration
        thin_dev = ThinDevice.create_from_pool(
            pool=self,
            thin_id=thin_id,
            size=size,
        )

        if activate and thin_dev.create(dm_name).failed:
            logger.error(f'Failed to activate thin device {dm_name}')
            # Clean up the thin device from pool
            self.message(0, f'"delete {thin_id}"')
            return None

        # Track the thin device
        self.thin_devices[thin_id] = thin_dev
        return thin_dev

    def delete_thin(self, thin_id: int, *, force: bool = False) -> CommandResult:
        """Delete a thin volume from the pool.

        Removes the thin device mapping and deletes it from the pool.
        """
        if not self.is_created:
            raise DmError('Pool must be created before deleting thin devices')

        # Remove the DM device if it's tracked and activated
        if thin_id in self.thin_devices:
            thin_dev = self.thin_devices[thin_id]
            if thin_dev.is_created:
                remove_result = thin_dev.remove(force=force)
                if remove_result.failed:
                    logger.error(f'Failed to remove thin device {thin_dev.dm_name}')
                    return remove_result

        # Delete thin from pool
        result = self.message(0, f'"delete {thin_id}"')
        if result.failed:
            logger.error(f'Failed to delete thin device {thin_id} from pool')
            return result

        # Remove from tracking
        self.thin_devices.pop(thin_id, None)
        logger.debug(f'Deleted thin device ID {thin_id} from pool {self.dm_name}')
        return result

    def create_snapshot(
        self,
        origin_id: int,
        snap_id: int,
        dm_name: str,
        *,
        size: int | None = None,
        activate: bool = True,
    ) -> ThinDevice | None:
        """Create a snapshot of an existing thin volume.

        The origin device should be suspended during snapshot creation.
        """
        if not self.is_created:
            logger.error('Pool must be created before creating snapshots')
            return None

        if snap_id in self.thin_devices:
            logger.error(f'Thin device with ID {snap_id} already exists in pool')
            return None

        # Determine size from origin if not specified
        if size is None:
            if origin_id in self.thin_devices:
                size = self.thin_devices[origin_id].size_sectors
            else:
                logger.error(f'Origin {origin_id} not tracked, size must be specified')
                return None

        # Send create_snap message to pool
        if self.message(0, f'"create_snap {snap_id} {origin_id}"').failed:
            logger.error(f'Failed to create snapshot {snap_id} of {origin_id}')
            return None

        logger.debug(f'Created snapshot ID {snap_id} of origin {origin_id} in pool {self.dm_name}')

        # Create ThinDevice configuration for snapshot
        snap_dev = ThinDevice.create_from_pool(
            pool=self,
            thin_id=snap_id,
            size=size,
        )

        if activate and snap_dev.create(dm_name).failed:
            logger.error(f'Failed to activate snapshot device {dm_name}')
            # Clean up the snapshot from pool
            self.message(0, f'"delete {snap_id}"')
            return None

        # Track the snapshot device
        self.thin_devices[snap_id] = snap_dev
        return snap_dev

    def remove(self, *, force: bool = False, retry: bool = False, deferred: bool = False) -> CommandResult:
        """Remove the thin pool, removing all child thin devices first."""
        # Remove all tracked thin devices first
        for thin_id in list(self.thin_devices.keys()):
            self.delete_thin(thin_id, force=force)

        # Remove the pool itself
        return super().remove(force=force, retry=retry, deferred=deferred)

create_snapshot(origin_id, snap_id, dm_name, *, size=None, activate=True)

Create a snapshot of an existing thin volume.

The origin device should be suspended during snapshot creation.

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

def create_snapshot(
    self,
    origin_id: int,
    snap_id: int,
    dm_name: str,
    *,
    size: int | None = None,
    activate: bool = True,
) -> ThinDevice | None:
    """Create a snapshot of an existing thin volume.

    The origin device should be suspended during snapshot creation.
    """
    if not self.is_created:
        logger.error('Pool must be created before creating snapshots')
        return None

    if snap_id in self.thin_devices:
        logger.error(f'Thin device with ID {snap_id} already exists in pool')
        return None

    # Determine size from origin if not specified
    if size is None:
        if origin_id in self.thin_devices:
            size = self.thin_devices[origin_id].size_sectors
        else:
            logger.error(f'Origin {origin_id} not tracked, size must be specified')
            return None

    # Send create_snap message to pool
    if self.message(0, f'"create_snap {snap_id} {origin_id}"').failed:
        logger.error(f'Failed to create snapshot {snap_id} of {origin_id}')
        return None

    logger.debug(f'Created snapshot ID {snap_id} of origin {origin_id} in pool {self.dm_name}')

    # Create ThinDevice configuration for snapshot
    snap_dev = ThinDevice.create_from_pool(
        pool=self,
        thin_id=snap_id,
        size=size,
    )

    if activate and snap_dev.create(dm_name).failed:
        logger.error(f'Failed to activate snapshot device {dm_name}')
        # Clean up the snapshot from pool
        self.message(0, f'"delete {snap_id}"')
        return None

    # Track the snapshot device
    self.thin_devices[snap_id] = snap_dev
    return snap_dev

create_thin(thin_id, size, dm_name, *, activate=True)

Create a new thin volume in the pool.

Sends 'create_thin' message to the pool and optionally activates the device.

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

def create_thin(
    self,
    thin_id: int,
    size: int,
    dm_name: str,
    *,
    activate: bool = True,
) -> ThinDevice | None:
    """Create a new thin volume in the pool.

    Sends 'create_thin' message to the pool and optionally activates the device.
    """
    if not self.is_created:
        logger.error('Pool must be created before creating thin devices')
        return None

    if thin_id in self.thin_devices:
        logger.error(f'Thin device with ID {thin_id} already exists in pool')
        return None

    # Send create_thin message to pool
    if self.message(0, f'"create_thin {thin_id}"').failed:
        logger.error(f'Failed to create thin device {thin_id} in pool')
        return None

    logger.debug(f'Created thin device ID {thin_id} in pool {self.dm_name}')

    # Create ThinDevice configuration
    thin_dev = ThinDevice.create_from_pool(
        pool=self,
        thin_id=thin_id,
        size=size,
    )

    if activate and thin_dev.create(dm_name).failed:
        logger.error(f'Failed to activate thin device {dm_name}')
        # Clean up the thin device from pool
        self.message(0, f'"delete {thin_id}"')
        return None

    # Track the thin device
    self.thin_devices[thin_id] = thin_dev
    return thin_dev

delete_thin(thin_id, *, force=False)

Delete a thin volume from the pool.

Removes the thin device mapping and deletes it from the pool.

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

def delete_thin(self, thin_id: int, *, force: bool = False) -> CommandResult:
    """Delete a thin volume from the pool.

    Removes the thin device mapping and deletes it from the pool.
    """
    if not self.is_created:
        raise DmError('Pool must be created before deleting thin devices')

    # Remove the DM device if it's tracked and activated
    if thin_id in self.thin_devices:
        thin_dev = self.thin_devices[thin_id]
        if thin_dev.is_created:
            remove_result = thin_dev.remove(force=force)
            if remove_result.failed:
                logger.error(f'Failed to remove thin device {thin_dev.dm_name}')
                return remove_result

    # Delete thin from pool
    result = self.message(0, f'"delete {thin_id}"')
    if result.failed:
        logger.error(f'Failed to delete thin device {thin_id} from pool')
        return result

    # Remove from tracking
    self.thin_devices.pop(thin_id, None)
    logger.debug(f'Deleted thin device ID {thin_id} from pool {self.dm_name}')
    return result

from_block_devices(metadata_device, data_device, block_size_sectors=128, low_water_mark=0, features=None, start=0, size=None) classmethod

Create ThinPoolDevice from BlockDevices.

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

@classmethod
def from_block_devices(
    cls,
    metadata_device: BlockDevice,
    data_device: BlockDevice,
    block_size_sectors: int = 128,
    low_water_mark: int = 0,
    features: list[str] | None = None,
    start: int = 0,
    size: int | None = None,
) -> ThinPoolDevice:
    """Create ThinPoolDevice from BlockDevices."""
    if features is None:
        features = ['skip_block_zeroing']

    metadata_id = cls._get_device_identifier(metadata_device)
    data_id = cls._get_device_identifier(data_device)

    if size is None and data_device.size is not None:
        size = data_device.size // data_device.sector_size

    # Build args: <metadata dev> <data dev> <block size> <low water mark> <# feature args> <feature args>
    args_parts = [
        metadata_id,
        data_id,
        str(block_size_sectors),
        str(low_water_mark),
        str(len(features)),
        *features,
    ]

    args = ' '.join(args_parts)
    if size is None:
        raise ValueError('size must be provided or data_device.size must be available')
    return cls(start=start, size_sectors=size, args=args)

remove(*, force=False, retry=False, deferred=False)

Remove the thin pool, removing all child thin devices first.

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

def remove(self, *, force: bool = False, retry: bool = False, deferred: bool = False) -> CommandResult:
    """Remove the thin pool, removing all child thin devices first."""
    # Remove all tracked thin devices first
    for thin_id in list(self.thin_devices.keys()):
        self.delete_thin(thin_id, force=force)

    # Remove the pool itself
    return super().remove(force=force, retry=retry, deferred=deferred)

sts.dm.zero

Device Mapper zero target.

ZeroDevice pydantic-model

Bases: DmDevice

Zero target -- returns zeros on read, discards writes.

Takes no target arguments.

Show JSON schema:

{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Zero target -- returns zeros on read, discards writes.\n\nTakes no target arguments.",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "start": {
      "default": 0,
      "title": "Start",
      "type": "integer"
    },
    "size_sectors": {
      "default": 0,
      "title": "Size Sectors",
      "type": "integer"
    },
    "args": {
      "default": "",
      "title": "Args",
      "type": "string"
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "target_type": {
      "default": "zero",
      "title": "Target Type",
      "type": "string"
    },
    "table": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Table"
    },
    "is_created": {
      "default": false,
      "title": "Is Created",
      "type": "boolean"
    }
  },
  "title": "ZeroDevice",
  "type": "object"
}

Fields:

• path (PathOrStr | None)
• name (str | None)
• size (int | None)
• model (str | None)
• blockdev_info (BlockdevInfo | None)
• lsblk_info (LsblkInfo | None)
• start (int)
• size_sectors (int)
• args (str)
• dm_name (str | None)
• table (str | None)
• is_created (bool)
• target_type (str)

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

class ZeroDevice(DmDevice):
    """Zero target -- returns zeros on read, discards writes.

    Takes no target arguments.
    """

    target_type: str = Field(default='zero', init=False)

    @classmethod
    def create_config(
        cls,
        start: int = 0,
        size: int | None = None,
    ) -> ZeroDevice:
        """Create ZeroDevice configuration."""
        if size is None:
            raise ValueError('Size must be specified for zero targets')

        return cls(start=start, size_sectors=size, args='')

create_config(start=0, size=None) classmethod

Create ZeroDevice configuration.

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

@classmethod
def create_config(
    cls,
    start: int = 0,
    size: int | None = None,
) -> ZeroDevice:
    """Create ZeroDevice configuration."""
    if size is None:
        raise ValueError('Size must be specified for zero targets')

    return cls(start=start, size_sectors=size, args='')