@classmethoddeffrom_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 storing cache metadata cache_device: Fast device for cached data (e.g., SSD) origin_device: Slow device with original data (e.g., HDD) block_size_sectors: Cache block size in sectors (default: 512 = 256KB) Must be between 64 (32KB) and 2097152 (1GB), multiple of 64 writethrough: Use writethrough mode (writes go to both devices) passthrough: Use passthrough mode (all I/O to origin) metadata2: Use version 2 metadata format no_discard_passdown: Don't pass discards to origin policy: Cache policy name (default: 'default') policy_args: Policy-specific key/value arguments start: Start sector in virtual device (default: 0) size_sectors: Size in sectors (default: origin device size) Returns: CacheDevice instance Example: ```python metadata = BlockDevice('/dev/sdb1') # Small metadata device cache = BlockDevice('/dev/nvme0n1') # Fast SSD cache origin = BlockDevice('/dev/sda') # Slow HDD origin cache_dev = CacheDevice.from_block_devices( metadata_device=metadata, cache_device=cache, origin_device=origin, block_size_sectors=1024, # 512KB blocks policy='smq', ) ``` """ifsize_sectorsisNoneandorigin_device.sizeisnotNone:size_sectors=origin_device.size//origin_device.sector_sizemetadata_id=cls._get_device_identifier(metadata_device)cache_id=cls._get_device_identifier(cache_device)origin_id=cls._get_device_identifier(origin_device)# Build feature listfeatures:list[str]=[]ifwritethrough:features.append('writethrough')ifpassthrough:features.append('passthrough')ifmetadata2:features.append('metadata2')ifno_discard_passdown:features.append('no_discard_passdown')# Build policy args listpolicy_args_list:list[str]=[]ifpolicy_args:forkey,valueinpolicy_args.items():policy_args_list.extend([key,str(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.append(policy)args_parts.append(str(len(policy_args_list)))args_parts.extend(policy_args_list)args=' '.join(args_parts)ifsize_sectorsisNone:raiseValueError('size_sectors must be provided or origin_device.size must be available')returncls(start=start,size_sectors=size_sectors,args=args)
Delays reads and/or writes and/or flushes and optionally maps them
to different devices. Useful for testing how applications handle slow devices.
Table line has to either have 3, 6 or 9 arguments:
3 args:
Apply offset and delay to read, write and flush operations on device
6 args:
Apply offset and delay to device for reads, also apply write_offset and
write_delay to write and flush operations on optionally different write_device
9 args:
Same as 6 arguments plus define flush_offset and flush_delay explicitly
on/with optionally different flush_device/flush_offset
Offsets are specified in sectors. Delays are specified in milliseconds.
Attributes:
Name
Type
Description
read_device
str | None
Device for read operations (parsed from args)
read_offset
int | None
Offset in read device sectors (parsed from args)
read_delay_ms
int | None
Delay for read operations in milliseconds (parsed from args)
write_device
str | None
Device for write operations (parsed from args, 6+ arg format)
write_offset
int | None
Offset in write device sectors (parsed from args, 6+ arg format)
write_delay_ms
int | None
Delay for write operations in milliseconds (parsed from args, 6+ arg format)
flush_device
str | None
Device for flush operations (parsed from args, 9 arg format)
flush_offset
int | None
Offset in flush device sectors (parsed from args, 9 arg format)
flush_delay_ms
int | None
Delay for flush operations in milliseconds (parsed from args, 9 arg format)
@classmethoddeffrom_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). This applies the same delay to read, write, and flush operations. Args: device: Source block device delay_ms: Delay in milliseconds for all operations start: Start sector in virtual device (default: 0) size_sectors: Size in sectors (default: device size in sectors) offset: Offset in source device sectors (default: 0) Returns: DelayDevice instance Example: ```python device = BlockDevice('/dev/sdb') delay = DelayDevice.from_block_device(device, delay_ms=100, size_sectors=1000000) delay.create('my-delay') ``` """ifsize_sectorsisNoneanddevice.sizeisnotNone:size_sectors=device.size//device.sector_sizedevice_id=cls._get_device_identifier(device)args=f'{device_id}{offset}{delay_ms}'ifsize_sectorsisNone:raiseValueError('size_sectors must be provided or device.size must be available')dev=cls(start=start,size_sectors=size_sectors,args=args)# Set attributes directly - we know the valuesdev.read_device=device_iddev.read_offset=offsetdev.read_delay_ms=delay_msdev.arg_format='3'returndev
@classmethoddeffrom_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 DelayTarget with separate read and write/flush devices (6 argument format). This allows different devices, offsets, and delays for read operations versus write and flush operations. Args: read_device: Device for read operations read_offset: Offset in read device sectors read_delay_ms: Delay for read operations in milliseconds write_device: Device for write and flush operations write_offset: Offset in write device sectors write_delay_ms: Delay for write and flush operations in milliseconds start: Start sector in virtual device (default: 0) size: Size in sectors (default: read device size in sectors) Returns: DelayTarget instance Example: ```python read_dev = BlockDevice('/dev/sdb') write_dev = BlockDevice('/dev/sdc') target = DelayTarget.from_block_devices_rw( read_dev, read_offset=2048, read_delay_ms=0, write_dev, write_offset=4096, write_delay_ms=400, size=1000000 ) str(target) '0 1000000 delay 8:16 2048 0 8:32 4096 400' ``` """ifsizeisNoneandread_device.sizeisnotNone:size=read_device.size//read_device.sector_sizeread_device_id=cls._get_device_identifier(read_device)write_device_id=cls._get_device_identifier(write_device)args=f'{read_device_id}{read_offset}{read_delay_ms}{write_device_id}{write_offset}{write_delay_ms}'ifsizeisNone:raiseValueError('size must be provided or read_device.size must be available')target=cls(start=start,size_sectors=size,args=args)# Set attributes directly - we know the valuestarget.read_device=read_device_idtarget.read_offset=read_offsettarget.read_delay_ms=read_delay_mstarget.write_device=write_device_idtarget.write_offset=write_offsettarget.write_delay_ms=write_delay_mstarget.arg_format='6'returntarget
@classmethoddeffrom_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 DelayTarget with separate read, write, and flush devices (9 argument format). This allows different devices, offsets, and delays for read, write, and flush operations separately. Args: read_device: Device for read operations read_offset: Offset in read device sectors read_delay_ms: Delay for read operations in milliseconds write_device: Device for write operations write_offset: Offset in write device sectors write_delay_ms: Delay for write operations in milliseconds flush_device: Device for flush operations flush_offset: Offset in flush device sectors flush_delay_ms: Delay for flush operations in milliseconds start: Start sector in virtual device (default: 0) size: Size in sectors (default: read device size in sectors) Returns: DelayTarget instance Example: ```python device = BlockDevice('/dev/sdb') target = DelayTarget.from_block_devices_rwf( device, read_offset=0, read_delay_ms=50, device, write_offset=0, write_delay_ms=100, device, flush_offset=0, flush_delay_ms=333, size=1000000 ) str(target) '0 1000000 delay 8:16 0 50 8:16 0 100 8:16 0 333' ``` """ifsizeisNoneandread_device.sizeisnotNone:size=read_device.size//read_device.sector_sizeread_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)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}')ifsizeisNone:raiseValueError('size must be provided or read_device.size must be available')target=cls(start=start,size_sectors=size,args=args)# Set attributes directly - we know the valuestarget.read_device=read_device_idtarget.read_offset=read_offsettarget.read_delay_ms=read_delay_mstarget.write_device=write_device_idtarget.write_offset=write_offsettarget.write_delay_ms=write_delay_mstarget.flush_device=flush_device_idtarget.flush_offset=flush_offsettarget.flush_delay_ms=flush_delay_mstarget.arg_format='9'returntarget
Parses a full table line (including start, size, and type) and
creates a DelayTarget instance. Useful for reconstructing targets
from existing devices.
@classmethoddeffrom_table_line(cls,table_line:str)->DelayDevice|None:"""Create DelayTarget from a dmsetup table line. Parses a full table line (including start, size, and type) and creates a DelayTarget instance. Useful for reconstructing targets from existing devices. Args: table_line: Full table line from 'dmsetup table' output Returns: DelayTarget instance or None if parsing fails Example: ```python line = '0 2097152 delay 8:16 0 100' target = DelayTarget.from_table_line(line) target.size # 2097152 ``` """parts=table_line.strip().split(None,3)iflen(parts)<4:logging.warning(f'Invalid table line: {table_line}')returnNonestart=int(parts[0])size=int(parts[1])target_type=parts[2]args=parts[3]iftarget_type!='delay':logging.warning(f'Not a delay target: {target_type}')returnNonetarget=cls(start=start,size_sectors=size,args=args)target.refresh()# Parse args to populate attributesreturntarget
A DM device can be in two states:
1. Configuration: has target config (start, size, args) but not yet created on system
2. Active: device exists on system, has path, dm_name, table, etc.
This class unifies target configuration and device management. Before calling
create(), the device is just a configuration. After create(), it becomes a
full block device with all BlockDevice functionality.
Device Mapper name (set after creation or when loading existing device)
None
name
str | None
Kernel device name like 'dm-0' (set after creation)
None
path
Path | str | None
Device path like '/dev/dm-0' (set after creation)
None
Example
# Create configurationdevice=LinearDevice.from_block_device(backing_dev,size=1000000)# Create on systemdevice.create('my-linear')# Now it's a full devicedevice.dm_device_path# '/dev/mapper/my-linear'device.suspend()device.resume()device.remove()
@dataclassclassDmDevice(BlockDevice):"""Base class for all Device Mapper devices. A DM device can be in two states: 1. Configuration: has target config (start, size, args) but not yet created on system 2. Active: device exists on system, has path, dm_name, table, etc. This class unifies target configuration and device management. Before calling create(), the device is just a configuration. After create(), it becomes a full block device with all BlockDevice functionality. Args: start: Start sector (where this target begins) size: Size in sectors (length of this target) args: Target-specific arguments (e.g. device paths, options) dm_name: Device Mapper name (set after creation or when loading existing device) name: Kernel device name like 'dm-0' (set after creation) path: Device path like '/dev/dm-0' (set after creation) Example: ```python # Create configuration device = LinearDevice.from_block_device(backing_dev, size=1000000) # Create on system device.create('my-linear') # Now it's a full device device.dm_device_path # '/dev/mapper/my-linear' device.suspend() device.resume() device.remove() ``` """# Target configuration (always available)start:int=0size_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 subclassestarget_type:str=field(init=False,default='')# Internal statetable:str|None=field(init=False,default=None,repr=False)is_created:bool=field(init=False,default=False,repr=False)# Class-level pathsDM_PATH:ClassVar[Path]=Path('/sys/class/block')DM_DEV_PATH:ClassVar[Path]=Path('/dev/mapper')def__post_init__(self)->None:"""Initialize Device Mapper device. If dm_name or path is provided, assumes device exists and initializes BlockDevice properties. Otherwise, stays in configuration-only state. """# Check if this is an existing device (has dm_name or path)ifself.dm_nameorself.path:# Set path based on dm_name if not providedifnotself.pathandself.dm_name:self.path=f'/dev/mapper/{self.dm_name}'elifnotself.pathandself.name:self.path=f'/dev/{self.name}'# Initialize BlockDevice (queries system)super().__post_init__()# Get device mapper name if not providedifnotself.dm_nameandself.name:result=run(f'dmsetup info -c --noheadings -o name {self.name}')ifresult.succeeded:self.dm_name=result.stdout.strip()# Load table from systemifself.dm_name:result=run(f'dmsetup table {self.dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()self.is_created=True@propertydeftype(self)->str:"""Get target type. Returns: Target type string (e.g. 'linear', 'delay', 'vdo') """ifself.target_type:returnself.target_type# Fallback: Remove 'Device' suffix from class name and convert to lowercasereturnself.__class__.__name__.lower().removesuffix('device')def__str__(self)->str:"""Return target table entry. Format: <start> <size> <type> <args> Used in dmsetup table commands. """returnf'{self.start}{self.size_sectors}{self.type}{self.args}'@propertydefdevice_path(self)->Path:"""Get path to device in sysfs. Returns: Path to device directory Raises: DeviceNotFoundError: If device does not exist """ifnotself.is_createdornotself.name:msg='Device not created yet'raiseDeviceNotFoundError(msg)path=self.DM_PATH/self.nameifnotpath.exists():msg=f'Device {self.name} not found'raiseDeviceNotFoundError(msg)returnpath@propertydefdevice_root_dir(self)->str:"""Get device mapper root directory. Returns: Device mapper root directory path (/dev/mapper) """returnstr(self.DM_DEV_PATH)@propertydefdm_device_path(self)->str|None:"""Get full device mapper device path. Returns: Full device path (e.g. '/dev/mapper/my-device') or None if not available """ifnotself.dm_name:returnNonereturnf'{self.device_root_dir}/{self.dm_name}'defcreate(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,)->bool:"""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: Name for the device mapper device uuid: Optional UUID for the device (can be used in place of name in commands) readonly: Set the table being loaded as read-only notable: Create device without loading any table readahead: Read ahead size in sectors, or 'auto', 'none', or '+N' for minimum addnodeoncreate: Ensure /dev/mapper node exists after create addnodeonresume: Ensure /dev/mapper node exists after resume (default with udev) Returns: True if successful, False otherwise Example: ```python device = DelayDevice.from_block_device(backing, delay_ms=100) if device.create('my-delay', uuid='my-uuid-123'): print(f'Created: {device.dm_device_path}') ``` """ifself.is_created:logging.warning(f'Device {self.dm_name} already created')returnTruecmd=f'dmsetup create {dm_name}'ifuuid:cmd+=f' --uuid {uuid}'ifreadonly:cmd+=' --readonly'ifaddnodeoncreate:cmd+=' --addnodeoncreate'ifaddnodeonresume:cmd+=' --addnodeonresume'ifreadaheadisnotNone:cmd+=f' --readahead {readahead}'ifnotable:cmd+=' --notable'else:# Build table from this device's configurationtable=str(self)cmd+=f' --table "{table}"'logging.info(f'Creating device {dm_name} with table: {table}')result=run(cmd)ifresult.failed:logging.error(f'Failed to create device {dm_name}: {result.stderr}')returnFalselogging.info(f'Successfully created device mapper device: {dm_name}')# Set dm_name and reinitialize as existing deviceself.dm_name=dm_nameself.path=f'/dev/mapper/{dm_name}'# Initialize BlockDevice properties from systemsuper().__post_init__()# Get kernel device nameifnotself.name:result=run(f'dmsetup info -c --noheadings -o name,blkdevname {dm_name}')ifresult.succeeded:parts=result.stdout.strip().split()iflen(parts)>=2:self.name=parts[1]# Load table from systemresult=run(f'dmsetup table {dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()self.is_created=TruereturnTruedef_parse_table(self)->None:"""Parse table and update target configuration from system. Subclasses should override this to parse target-specific attributes. """ifnotself.table:returnparts=self.table.strip().split(None,3)iflen(parts)>=4:self.start=int(parts[0])self.size_sectors=int(parts[1])# parts[2] is the target typeself.args=parts[3]# Call refresh if available (subclasses implement this for target-specific parsing)ifhasattr(self,'refresh')andcallable(getattr(self,'refresh',None)):self.refresh()defrefresh(self)->None:"""Refresh device state from system. Re-reads the device table and status from the system. Subclasses should override to parse target-specific attributes. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:return# Refresh table from systemresult=run(f'dmsetup table {self.dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()defget_status(self,*,target_type:str|None=None,noflush:bool=False,)->str|None:"""Get device status from dmsetup. Outputs status information for each of the device's targets. Args: target_type: Only display information for the specified target type noflush: Do not commit thin-pool metadata before reporting statistics Returns: Status string or None if not available """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup status {self.dm_name}'iftarget_type:cmd+=f' --target {target_type}'ifnoflush:cmd+=' --noflush'result=run(cmd)ifresult.failed:logging.error(f'Failed to get status for {self.dm_name}')returnNonereturnresult.stdout.strip()defsuspend(self,*,nolockfs:bool=False,noflush:bool=False,)->bool:"""Suspend device. Suspends I/O to the device. Any I/O that has already been mapped by the device but has not yet completed will be flushed. Any further I/O to that device will be postponed for as long as the device is suspended. Args: nolockfs: Do not attempt to synchronize filesystem (skip fsfreeze) noflush: Let outstanding I/O remain unflushed (requires target support) Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup suspend {self.dm_name}'ifnolockfs:cmd+=' --nolockfs'ifnoflush:cmd+=' --noflush'result=run(cmd)ifresult.failed:logging.error('Failed to suspend device')returnFalsereturnTruedefresume(self,*,addnodeoncreate:bool=False,addnodeonresume:bool=False,noflush:bool=False,nolockfs:bool=False,readahead:str|int|None=None,)->bool:"""Resume device. Un-suspends a device. If an inactive table has been loaded, it becomes live. Postponed I/O then gets re-queued for processing. Args: addnodeoncreate: Ensure /dev/mapper node exists after create addnodeonresume: Ensure /dev/mapper node exists after resume (default with udev) noflush: Do not flush outstanding I/O nolockfs: Do not attempt to synchronize filesystem readahead: Read ahead size in sectors, or 'auto', 'none', or '+N' for minimum Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup resume {self.dm_name}'ifaddnodeoncreate:cmd+=' --addnodeoncreate'ifaddnodeonresume:cmd+=' --addnodeonresume'ifnoflush:cmd+=' --noflush'ifnolockfs:cmd+=' --nolockfs'ifreadaheadisnotNone:cmd+=f' --readahead {readahead}'result=run(cmd)ifresult.failed:logging.error('Failed to resume device')returnFalsereturnTruedefremove(self,*,force:bool=False,retry:bool=False,deferred:bool=False,)->bool:"""Remove device. Removes the device mapping. The underlying devices are unaffected. Open devices cannot be removed, but --force will replace the table with one that fails all I/O. --deferred enables deferred removal when the device is in use. 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: Enable deferred removal - device removed when last user closes it Returns: True if successful, False otherwise Note: Do NOT combine --force and --udevcookie as this may cause nondeterministic results. """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup remove {self.dm_name}'ifforce:cmd+=' --force'ifretry:cmd+=' --retry'ifdeferred:cmd+=' --deferred'result=run(cmd)ifresult.failed:logging.error('Failed to remove device')returnFalseself.is_created=FalsereturnTruedefclear(self)->bool:"""Clear the inactive table slot. Destroys the table in the inactive table slot for the device. This is useful when you want to abort a table load before resuming. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup clear {self.dm_name}')ifresult.failed:logging.error(f'Failed to clear inactive table for {self.dm_name}')returnFalsereturnTruedefdeps(self,*,output_format:str='devno')->list[str]:"""Get device dependencies. Outputs a list of devices referenced by the live table for this device. Args: output_format: Output format for device names: - 'devno': Major and minor pair (default) - 'blkdevname': Block device name - 'devname': Map name for DM devices, blkdevname otherwise Returns: List of device identifiers (format depends on output_format) """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')return[]result=run(f'dmsetup deps -o {output_format}{self.dm_name}')ifresult.failed:logging.error(f'Failed to get dependencies for {self.dm_name}')return[]# Parse output: "1 dependencies : (major, minor) ..."output=result.stdout.strip()deps=[]if':'inoutput:deps_part=output.split(':',1)[1].strip()# Extract device identifiers from the outputifoutput_format=='devno':# Format: (major, minor) or (major:minor)matches=re.findall(r'\((\d+[,:]\s*\d+)\)',deps_part)deps=[m.replace(' ','').replace(',',':')forminmatches]else:# Format: (device_name)matches=re.findall(r'\(([^)]+)\)',deps_part)deps=matchesreturndepsdefinfo(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. Outputs information about the device in various formats. Args: columns: Display output in columns rather than Field: Value lines noheadings: Suppress the headings line when using columnar output fields: Comma-separated list of fields to display (name, major, minor, attr, open, segments, events, uuid) separator: Separator for columnar output sort_fields: Fields to sort by (prefix with '-' for reverse) nameprefixes: Add DM_ prefix plus field name to output Returns: If columns=False: Dictionary of field names to values If columns=True: Raw output string None if device not found or error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup info {self.dm_name}'ifcolumns:cmd+=' --columns'ifnoheadings:cmd+=' --noheadings'iffields:cmd+=f' -o {fields}'ifseparator:cmd+=f' --separator "{separator}"'ifsort_fields:cmd+=f' --sort {sort_fields}'ifnameprefixes:cmd+=' --nameprefixes'result=run(cmd)ifresult.failed:logging.error(f'Failed to get info for {self.dm_name}')returnNoneoutput=result.stdout.strip()ifcolumns:returnoutput# Parse Field: Value format into dictionaryinfo_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)info_dict[key.strip()]=value.strip()returninfo_dictdefmessage(self,sector:int,message:str)->bool:"""Send message to device target. Send a message to the target at the specified sector. Use sector 0 if the sector is not relevant to the target. Args: sector: Sector number (use 0 if not needed) message: Message string to send to the target Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup message {self.dm_name}{sector}{message}')ifresult.failed:logging.error(f'Failed to send message to {self.dm_name}: {result.stderr}')returnFalsereturnTruedefrename(self,new_name:str)->bool:"""Rename the device. Args: new_name: New name for the device Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup rename {self.dm_name}{new_name}')ifresult.failed:logging.error(f'Failed to rename {self.dm_name} to {new_name}')returnFalse# Update internal stateold_name=self.dm_nameself.dm_name=new_nameself.path=f'/dev/mapper/{new_name}'logging.info(f'Successfully renamed device from {old_name} to {new_name}')returnTruedefset_uuid(self,uuid:str)->bool:"""Set the UUID of a device that was created without one. After a UUID has been set it cannot be changed. Args: uuid: UUID to set for the device Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup rename {self.dm_name} --setuuid {uuid}')ifresult.failed:logging.error(f'Failed to set UUID for {self.dm_name}: {result.stderr}')returnFalselogging.info(f'Successfully set UUID {uuid} for device {self.dm_name}')returnTruedefreload_table(self,new_table:str)->bool:"""Reload device table. Suspends the device, loads a new table, and resumes. Args: new_table: New table string to load Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalse# Suspend deviceifnotself.suspend():returnFalse# Load new tableresult=run(f'dmsetup load {self.dm_name} "{new_table}"')ifresult.failed:logging.error(f'Failed to load new table: {result.stderr}')self.resume()returnFalse# Resume with new tableifnotself.resume():returnFalse# Refresh from systemself.refresh()logging.info(f'Successfully reloaded table for {self.dm_name}')returnTruedefload(self,table:str|None=None,table_file:str|None=None,)->bool:"""Load table into the inactive table slot. Loads a table into the inactive table slot for the device. Use resume() to make the inactive table live. Args: table: Table string to load table_file: Path to file containing the table Returns: True if successful, False otherwise Note: Either table or table_file must be provided, but not both. """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseiftableandtable_file:raiseValueError('Cannot specify both table and table_file')iftable:result=run(f'dmsetup load {self.dm_name} --table "{table}"')eliftable_file:result=run(f'dmsetup load {self.dm_name}{table_file}')else:raiseValueError('Either table or table_file must be provided')ifresult.failed:logging.error(f'Failed to load table for {self.dm_name}: {result.stderr}')returnFalsereturnTruedefget_table(self,*,concise:bool=False,target_type:str|None=None,showkeys:bool=False,)->str|None:"""Get the current device table. Outputs the current table for the device in a format that can be fed back using create or load commands. Args: concise: Output in concise format (name,uuid,minor,flags,table) target_type: Only show information for the specified target type showkeys: Show real encryption keys (for crypt/integrity targets) Returns: Table string or None if error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup table {self.dm_name}'ifconcise:cmd+=' --concise'iftarget_type:cmd+=f' --target {target_type}'ifshowkeys:cmd+=' --showkeys'result=run(cmd)ifresult.failed:logging.error(f'Failed to get table for {self.dm_name}')returnNonereturnresult.stdout.strip()defwipe_table(self,*,force:bool=False,noflush:bool=False,nolockfs:bool=False,)->bool:"""Wipe device table. Wait for any I/O in-flight through the device to complete, then replace the table with a new table that fails any new I/O sent to the device. If successful, this should release any devices held open by the device's table(s). Args: force: Force the operation noflush: Do not flush outstanding I/O nolockfs: Do not attempt to synchronize filesystem Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup wipe_table {self.dm_name}'ifforce:cmd+=' --force'ifnoflush:cmd+=' --noflush'ifnolockfs:cmd+=' --nolockfs'result=run(cmd)ifresult.failed:logging.error(f'Failed to wipe table for {self.dm_name}: {result.stderr}')returnFalsereturnTruedefmknodes(self)->bool:"""Ensure /dev/mapper node is correct. Ensures that the node in /dev/mapper for this device is correct. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup mknodes {self.dm_name}')ifresult.failed:logging.error(f'Failed to mknodes for {self.dm_name}')returnFalsereturnTruedefsetgeometry(self,cylinders:int,heads:int,sectors:int,start:int,)->bool:"""Set the device geometry. Sets the device geometry to C/H/S format. Args: cylinders: Number of cylinders heads: Number of heads sectors: Sectors per track start: Start sector Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup setgeometry {self.dm_name}{cylinders}{heads}{sectors}{start}')ifresult.failed:logging.error(f'Failed to set geometry for {self.dm_name}: {result.stderr}')returnFalsereturnTruedefwait(self,event_nr:int|None=None,*,noflush:bool=False,)->int|None:"""Wait for device event. Sleeps until the event counter for the device exceeds event_nr. Args: event_nr: Event number to wait for (waits until counter exceeds this) noflush: Do not commit thin-pool metadata before reporting Returns: Current event number after waiting, or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup wait {self.dm_name}'ifnoflush:cmd+=' --noflush'ifevent_nrisnotNone:cmd+=f' {event_nr}'result=run(cmd)ifresult.failed:logging.error(f'Failed to wait for {self.dm_name}')returnNone# Parse event number from outputtry:returnint(result.stdout.strip())exceptValueError:returnNonedefmeasure(self)->str|None:"""Show IMA measurement data. Shows the data that the device would report to the IMA subsystem if a measurement was triggered. This is for debugging and does not actually trigger a measurement. Returns: Measurement data string or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNoneresult=run(f'dmsetup measure {self.dm_name}')ifresult.failed:logging.error(f'Failed to measure {self.dm_name}')returnNonereturnresult.stdout.strip()defmangle(self)->bool:"""Mangle device name. Ensures the device name and UUID contains only whitelisted characters (supported by udev) and renames if necessary. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup mangle {self.dm_name}')ifresult.failed:logging.error(f'Failed to mangle {self.dm_name}')returnFalsereturnTruedefsplitname(self,subsystem:str='LVM')->dict[str,str]|None:"""Split device name into subsystem constituents. Splits the device name into its subsystem components. LVM generates device names by concatenating Volume Group, Logical Volume, and any internal Layer with a hyphen separator. Args: subsystem: Subsystem to use for splitting (default: LVM) Returns: Dictionary with split name components or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNoneresult=run(f'dmsetup splitname {self.dm_name}{subsystem}')ifresult.failed:logging.error(f'Failed to split name {self.dm_name}')returnNone# Parse output into dictionaryoutput=result.stdout.strip()name_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)name_dict[key.strip()]=value.strip()returnname_dict@staticmethoddefversion()->dict[str,str]|None:"""Get dmsetup and driver version information. Returns: Dictionary with 'library' and 'driver' version strings, or None on error """result=run('dmsetup version')ifresult.failed:logging.error('Failed to get dmsetup version')returnNoneoutput=result.stdout.strip()version_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)version_dict[key.strip().lower().replace(' ','_')]=value.strip()returnversion_dict@staticmethoddeftargets()->list[dict[str,str]]:"""Get list of available device mapper targets. Displays the names and versions of currently-loaded targets. Returns: List of dictionaries with 'name' and 'version' keys """result=run('dmsetup targets')ifresult.failed:logging.error('Failed to get targets')return[]targets_list:list[dict[str,str]]=[]forlineinresult.stdout.strip().splitlines():parts=line.strip().split()iflen(parts)>=2:targets_list.append({'name':parts[0],'version':parts[1],})returntargets_list@staticmethoddefudevcreatecookie()->str|None:"""Create a new udev synchronization cookie. Creates a new cookie to synchronize actions with udev processing. The cookie can be used with --udevcookie option or exported as DM_UDEV_COOKIE environment variable. Returns: Cookie value string or None on error Note: The cookie is a system-wide semaphore that needs to be cleaned up explicitly by calling udevreleasecookie(). """result=run('dmsetup udevcreatecookie')ifresult.failed:logging.error('Failed to create udev cookie')returnNonereturnresult.stdout.strip()@staticmethoddefudevreleasecookie(cookie:str|None=None)->bool:"""Release a udev synchronization cookie. Waits for all pending udev processing bound to the cookie value and cleans up the cookie with underlying semaphore. Args: cookie: Cookie value to release. If not provided, uses DM_UDEV_COOKIE environment variable. Returns: True if successful, False otherwise """cmd='dmsetup udevreleasecookie'ifcookie:cmd+=f' {cookie}'result=run(cmd)ifresult.failed:logging.error('Failed to release udev cookie')returnFalsereturnTrue@staticmethoddefudevcomplete(cookie:str)->bool:"""Signal udev processing completion. Wakes any processes that are waiting for udev to complete processing the specified cookie. Args: cookie: Cookie value to signal completion for Returns: True if successful, False otherwise """result=run(f'dmsetup udevcomplete {cookie}')ifresult.failed:logging.error(f'Failed to complete udev cookie {cookie}')returnFalsereturnTrue@staticmethoddefudevcomplete_all(age_in_minutes:int|None=None)->bool:"""Remove all old udev cookies. Removes all cookies older than the specified number of minutes. Any process waiting on a cookie will be resumed immediately. Args: age_in_minutes: Remove cookies older than this many minutes Returns: True if successful, False otherwise """cmd='dmsetup udevcomplete_all'ifage_in_minutesisnotNone:cmd+=f' {age_in_minutes}'result=run(cmd)ifresult.failed:logging.error('Failed to complete all udev cookies')returnFalsereturnTrue@staticmethoddefudevcookie()->list[str]:"""List all existing udev cookies. Cookies are system-wide semaphores with keys prefixed by two predefined bytes (0x0D4D). Returns: List of cookie values """result=run('dmsetup udevcookie')ifresult.failed:logging.error('Failed to list udev cookies')return[]returnresult.stdout.strip().splitlines()@staticmethoddefudevflags(cookie:str)->dict[str,str]:"""Parse udev control flags from cookie. Parses the cookie value and extracts any udev control flags encoded. The output is in environment key format suitable for use in udev rules. Args: cookie: Cookie value to parse Returns: Dictionary of flag names to values (typically '1') """result=run(f'dmsetup udevflags {cookie}')ifresult.failed:logging.error(f'Failed to get udev flags for cookie {cookie}')return{}flags:dict[str,str]={}forlineinresult.stdout.strip().splitlines():if'='inline:key,value=line.split('=',1)flags[key.strip()]=value.strip().strip('\'"')returnflags@classmethoddefremove_all(cls,*,force:bool=False,deferred:bool=False)->bool:"""Remove all device mapper devices. Attempts to remove all device definitions (reset the driver). This also runs mknodes afterwards. Use with care! Args: force: Replace tables with one that fails all I/O before removing deferred: Enable deferred removal for open devices Returns: True if successful, False otherwise Warning: Open devices cannot be removed unless force or deferred is used. """cmd='dmsetup remove_all'ifforce:cmd+=' --force'ifdeferred:cmd+=' --deferred'result=run(cmd)ifresult.failed:logging.error('Failed to remove all devices')returnFalsereturnTrue@classmethoddefmknodes_all(cls)->bool:"""Ensure all /dev/mapper nodes are correct. Ensures that all nodes in /dev/mapper correspond to mapped devices currently loaded by the device-mapper kernel driver, adding, changing, or removing nodes as necessary. Returns: True if successful, False otherwise """result=run('dmsetup mknodes')ifresult.failed:logging.error('Failed to run mknodes')returnFalsereturnTrue@classmethoddefls(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. Lists device names with optional filtering and formatting. Args: target_type: Only list devices with at least one target of this type output_format: Output format for device names: - 'devno': Major and minor pair (default) - 'blkdevname': Block device name - 'devname': Map name for DM devices tree: Display dependencies between devices as a tree tree_options: Comma-separated tree display options: - device/nodevice, blkdevname, active, open, rw, uuid - ascii, utf, vt100, compact, inverted, notrunc exec_cmd: Execute command for each device (device name appended) Returns: If tree=True: Tree output as string Otherwise: List of device names """cmd='dmsetup ls'iftarget_type:cmd+=f' --target {target_type}'ifoutput_format!='devno':cmd+=f' -o {output_format}'iftree:cmd+=' --tree'iftree_options:cmd+=f' -o {tree_options}'ifexec_cmd:cmd+=f' --exec "{exec_cmd}"'result=run(cmd)ifresult.failed:logging.warning('No Device Mapper devices found')return[]ifnottreeelse''output=result.stdout.strip()iftreeorexec_cmd:returnoutput# Parse device names from outputdevices=[]forlineinoutput.splitlines():ifline.strip():parts=line.strip().split()ifparts:devices.append(parts[0])returndevices@classmethoddefget_all(cls)->Sequence[DmDevice]:"""Get list of all Device Mapper devices. Returns: List of DmDevice instances """devices=[]result=run('dmsetup ls')ifresult.failed:logging.warning('No Device Mapper devices found')return[]forlineinresult.stdout.splitlines():try:stripped_line=line.strip()ifnotstripped_line:continueparts=stripped_line.split()iflen(parts)<2:continuedm_name=parts[0]dev_id=parts[1].strip('()')major,minor=dev_id.split(':')# Get kernel device name (dm-N)result=run(f'ls -l /dev/dm-* | grep "{major}, *{minor}"')ifresult.failed:continuename=result.stdout.split('/')[-1].strip()devices.append(cls(dm_name=dm_name,name=name,path=f'/dev/{name}'))except(ValueError,DeviceError):logging.exception('Failed to parse device info')continuereturndevices@classmethoddefcreate_concise(cls,concise_spec:str)->bool:"""Create one or more devices from concise device specification. Each device is specified by a comma-separated list: name,uuid,minor,flags,table[,table]... Multiple devices are separated by semicolons. Args: concise_spec: Concise device specification string Format: <name>,<uuid>,<minor>,<flags>,<table>[,<table>+][;<next device>...] - name: Device name - uuid: Device UUID (optional, can be empty) - minor: Minor number (optional, can be empty for auto-assignment) - flags: 'ro' for read-only or 'rw' for read-write (default) - table: One or more table lines (comma-separated if multiple) Returns: True if successful, False otherwise Example: ```python # Single linear device DmDevice.create_concise('test-linear,,,rw,0 2097152 linear /dev/loop0 0') # Two devices at once DmDevice.create_concise('dev1,,,,0 2097152 linear /dev/loop0 0;dev2,,,,0 2097152 linear /dev/loop1 0') ``` Note: Use backslash to escape commas or semicolons in names or tables. """result=run(f'dmsetup create --concise "{concise_spec}"')ifresult.failed:logging.error(f'Failed to create devices from concise spec: {result.stderr}')returnFalsereturnTrue@staticmethoddefhelp(*,columns:bool=False)->str|None:"""Get dmsetup command help. Outputs a summary of the commands available. Args: columns: Include the list of report fields in the output Returns: Help text or None on error """cmd='dmsetup help'ifcolumns:cmd+=' --columns'result=run(cmd)# Note: dmsetup help returns exit code 0 but writes to stderroutput=result.stdout.strip()orresult.stderr.strip()returnoutputifoutputelseNone@classmethoddefget_by_name(cls,dm_name:str)->DmDevice|None:"""Get Device Mapper device by name. Args: dm_name: Device Mapper name (e.g. 'my-device') Returns: DmDevice instance or None if not found """ifnotdm_name:raiseValueError('Device Mapper name required')# Check if device existsresult=run(f'dmsetup info {dm_name}')ifresult.failed:returnNonetry:returncls(dm_name=dm_name)exceptDeviceError:returnNone@staticmethoddef_get_device_identifier(device:BlockDevice)->str:"""Get device identifier for device mapper. Args: device: BlockDevice instance Returns: Device identifier (major:minor format) Raises: DeviceError: If device identifier cannot be determined """device_id=device.device_idifdevice_id:returndevice_idraiseDeviceError(f'No device ID available for {device.path}')
def__post_init__(self)->None:"""Initialize Device Mapper device. If dm_name or path is provided, assumes device exists and initializes BlockDevice properties. Otherwise, stays in configuration-only state. """# Check if this is an existing device (has dm_name or path)ifself.dm_nameorself.path:# Set path based on dm_name if not providedifnotself.pathandself.dm_name:self.path=f'/dev/mapper/{self.dm_name}'elifnotself.pathandself.name:self.path=f'/dev/{self.name}'# Initialize BlockDevice (queries system)super().__post_init__()# Get device mapper name if not providedifnotself.dm_nameandself.name:result=run(f'dmsetup info -c --noheadings -o name {self.name}')ifresult.succeeded:self.dm_name=result.stdout.strip()# Load table from systemifself.dm_name:result=run(f'dmsetup table {self.dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()self.is_created=True
defclear(self)->bool:"""Clear the inactive table slot. Destroys the table in the inactive table slot for the device. This is useful when you want to abort a table load before resuming. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup clear {self.dm_name}')ifresult.failed:logging.error(f'Failed to clear inactive table for {self.dm_name}')returnFalsereturnTrue
defcreate(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,)->bool:"""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: Name for the device mapper device uuid: Optional UUID for the device (can be used in place of name in commands) readonly: Set the table being loaded as read-only notable: Create device without loading any table readahead: Read ahead size in sectors, or 'auto', 'none', or '+N' for minimum addnodeoncreate: Ensure /dev/mapper node exists after create addnodeonresume: Ensure /dev/mapper node exists after resume (default with udev) Returns: True if successful, False otherwise Example: ```python device = DelayDevice.from_block_device(backing, delay_ms=100) if device.create('my-delay', uuid='my-uuid-123'): print(f'Created: {device.dm_device_path}') ``` """ifself.is_created:logging.warning(f'Device {self.dm_name} already created')returnTruecmd=f'dmsetup create {dm_name}'ifuuid:cmd+=f' --uuid {uuid}'ifreadonly:cmd+=' --readonly'ifaddnodeoncreate:cmd+=' --addnodeoncreate'ifaddnodeonresume:cmd+=' --addnodeonresume'ifreadaheadisnotNone:cmd+=f' --readahead {readahead}'ifnotable:cmd+=' --notable'else:# Build table from this device's configurationtable=str(self)cmd+=f' --table "{table}"'logging.info(f'Creating device {dm_name} with table: {table}')result=run(cmd)ifresult.failed:logging.error(f'Failed to create device {dm_name}: {result.stderr}')returnFalselogging.info(f'Successfully created device mapper device: {dm_name}')# Set dm_name and reinitialize as existing deviceself.dm_name=dm_nameself.path=f'/dev/mapper/{dm_name}'# Initialize BlockDevice properties from systemsuper().__post_init__()# Get kernel device nameifnotself.name:result=run(f'dmsetup info -c --noheadings -o name,blkdevname {dm_name}')ifresult.succeeded:parts=result.stdout.strip().split()iflen(parts)>=2:self.name=parts[1]# Load table from systemresult=run(f'dmsetup table {dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()self.is_created=TruereturnTrue
Create one or more devices from concise device specification.
Each device is specified by a comma-separated list:
name,uuid,minor,flags,table[,table]...
Multiple devices are separated by semicolons.
Parameters:
Name
Type
Description
Default
concise_spec
str
Concise device specification string
Format: ,,,,
[,
+][;...]
- name: Device name
- uuid: Device UUID (optional, can be empty)
- minor: Minor number (optional, can be empty for auto-assignment)
- flags: 'ro' for read-only or 'rw' for read-write (default)
- table: One or more table lines (comma-separated if multiple)
required
Returns:
Type
Description
bool
True if successful, False otherwise
Example
# Single linear deviceDmDevice.create_concise('test-linear,,,rw,0 2097152 linear /dev/loop0 0')# Two devices at onceDmDevice.create_concise('dev1,,,,0 2097152 linear /dev/loop0 0;dev2,,,,0 2097152 linear /dev/loop1 0')
Note
Use backslash to escape commas or semicolons in names or tables.
@classmethoddefcreate_concise(cls,concise_spec:str)->bool:"""Create one or more devices from concise device specification. Each device is specified by a comma-separated list: name,uuid,minor,flags,table[,table]... Multiple devices are separated by semicolons. Args: concise_spec: Concise device specification string Format: <name>,<uuid>,<minor>,<flags>,<table>[,<table>+][;<next device>...] - name: Device name - uuid: Device UUID (optional, can be empty) - minor: Minor number (optional, can be empty for auto-assignment) - flags: 'ro' for read-only or 'rw' for read-write (default) - table: One or more table lines (comma-separated if multiple) Returns: True if successful, False otherwise Example: ```python # Single linear device DmDevice.create_concise('test-linear,,,rw,0 2097152 linear /dev/loop0 0') # Two devices at once DmDevice.create_concise('dev1,,,,0 2097152 linear /dev/loop0 0;dev2,,,,0 2097152 linear /dev/loop1 0') ``` Note: Use backslash to escape commas or semicolons in names or tables. """result=run(f'dmsetup create --concise "{concise_spec}"')ifresult.failed:logging.error(f'Failed to create devices from concise spec: {result.stderr}')returnFalsereturnTrue
Outputs a list of devices referenced by the live table for this device.
Parameters:
Name
Type
Description
Default
output_format
str
Output format for device names:
- 'devno': Major and minor pair (default)
- 'blkdevname': Block device name
- 'devname': Map name for DM devices, blkdevname otherwise
'devno'
Returns:
Type
Description
list[str]
List of device identifiers (format depends on output_format)
defdeps(self,*,output_format:str='devno')->list[str]:"""Get device dependencies. Outputs a list of devices referenced by the live table for this device. Args: output_format: Output format for device names: - 'devno': Major and minor pair (default) - 'blkdevname': Block device name - 'devname': Map name for DM devices, blkdevname otherwise Returns: List of device identifiers (format depends on output_format) """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')return[]result=run(f'dmsetup deps -o {output_format}{self.dm_name}')ifresult.failed:logging.error(f'Failed to get dependencies for {self.dm_name}')return[]# Parse output: "1 dependencies : (major, minor) ..."output=result.stdout.strip()deps=[]if':'inoutput:deps_part=output.split(':',1)[1].strip()# Extract device identifiers from the outputifoutput_format=='devno':# Format: (major, minor) or (major:minor)matches=re.findall(r'\((\d+[,:]\s*\d+)\)',deps_part)deps=[m.replace(' ','').replace(',',':')forminmatches]else:# Format: (device_name)matches=re.findall(r'\(([^)]+)\)',deps_part)deps=matchesreturndeps
@classmethoddefget_all(cls)->Sequence[DmDevice]:"""Get list of all Device Mapper devices. Returns: List of DmDevice instances """devices=[]result=run('dmsetup ls')ifresult.failed:logging.warning('No Device Mapper devices found')return[]forlineinresult.stdout.splitlines():try:stripped_line=line.strip()ifnotstripped_line:continueparts=stripped_line.split()iflen(parts)<2:continuedm_name=parts[0]dev_id=parts[1].strip('()')major,minor=dev_id.split(':')# Get kernel device name (dm-N)result=run(f'ls -l /dev/dm-* | grep "{major}, *{minor}"')ifresult.failed:continuename=result.stdout.split('/')[-1].strip()devices.append(cls(dm_name=dm_name,name=name,path=f'/dev/{name}'))except(ValueError,DeviceError):logging.exception('Failed to parse device info')continuereturndevices
@classmethoddefget_by_name(cls,dm_name:str)->DmDevice|None:"""Get Device Mapper device by name. Args: dm_name: Device Mapper name (e.g. 'my-device') Returns: DmDevice instance or None if not found """ifnotdm_name:raiseValueError('Device Mapper name required')# Check if device existsresult=run(f'dmsetup info {dm_name}')ifresult.failed:returnNonetry:returncls(dm_name=dm_name)exceptDeviceError:returnNone
defget_status(self,*,target_type:str|None=None,noflush:bool=False,)->str|None:"""Get device status from dmsetup. Outputs status information for each of the device's targets. Args: target_type: Only display information for the specified target type noflush: Do not commit thin-pool metadata before reporting statistics Returns: Status string or None if not available """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup status {self.dm_name}'iftarget_type:cmd+=f' --target {target_type}'ifnoflush:cmd+=' --noflush'result=run(cmd)ifresult.failed:logging.error(f'Failed to get status for {self.dm_name}')returnNonereturnresult.stdout.strip()
defget_table(self,*,concise:bool=False,target_type:str|None=None,showkeys:bool=False,)->str|None:"""Get the current device table. Outputs the current table for the device in a format that can be fed back using create or load commands. Args: concise: Output in concise format (name,uuid,minor,flags,table) target_type: Only show information for the specified target type showkeys: Show real encryption keys (for crypt/integrity targets) Returns: Table string or None if error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup table {self.dm_name}'ifconcise:cmd+=' --concise'iftarget_type:cmd+=f' --target {target_type}'ifshowkeys:cmd+=' --showkeys'result=run(cmd)ifresult.failed:logging.error(f'Failed to get table for {self.dm_name}')returnNonereturnresult.stdout.strip()
@staticmethoddefhelp(*,columns:bool=False)->str|None:"""Get dmsetup command help. Outputs a summary of the commands available. Args: columns: Include the list of report fields in the output Returns: Help text or None on error """cmd='dmsetup help'ifcolumns:cmd+=' --columns'result=run(cmd)# Note: dmsetup help returns exit code 0 but writes to stderroutput=result.stdout.strip()orresult.stderr.strip()returnoutputifoutputelseNone
definfo(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. Outputs information about the device in various formats. Args: columns: Display output in columns rather than Field: Value lines noheadings: Suppress the headings line when using columnar output fields: Comma-separated list of fields to display (name, major, minor, attr, open, segments, events, uuid) separator: Separator for columnar output sort_fields: Fields to sort by (prefix with '-' for reverse) nameprefixes: Add DM_ prefix plus field name to output Returns: If columns=False: Dictionary of field names to values If columns=True: Raw output string None if device not found or error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup info {self.dm_name}'ifcolumns:cmd+=' --columns'ifnoheadings:cmd+=' --noheadings'iffields:cmd+=f' -o {fields}'ifseparator:cmd+=f' --separator "{separator}"'ifsort_fields:cmd+=f' --sort {sort_fields}'ifnameprefixes:cmd+=' --nameprefixes'result=run(cmd)ifresult.failed:logging.error(f'Failed to get info for {self.dm_name}')returnNoneoutput=result.stdout.strip()ifcolumns:returnoutput# Parse Field: Value format into dictionaryinfo_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)info_dict[key.strip()]=value.strip()returninfo_dict
defload(self,table:str|None=None,table_file:str|None=None,)->bool:"""Load table into the inactive table slot. Loads a table into the inactive table slot for the device. Use resume() to make the inactive table live. Args: table: Table string to load table_file: Path to file containing the table Returns: True if successful, False otherwise Note: Either table or table_file must be provided, but not both. """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseiftableandtable_file:raiseValueError('Cannot specify both table and table_file')iftable:result=run(f'dmsetup load {self.dm_name} --table "{table}"')eliftable_file:result=run(f'dmsetup load {self.dm_name}{table_file}')else:raiseValueError('Either table or table_file must be provided')ifresult.failed:logging.error(f'Failed to load table for {self.dm_name}: {result.stderr}')returnFalsereturnTrue
@classmethoddefls(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. Lists device names with optional filtering and formatting. Args: target_type: Only list devices with at least one target of this type output_format: Output format for device names: - 'devno': Major and minor pair (default) - 'blkdevname': Block device name - 'devname': Map name for DM devices tree: Display dependencies between devices as a tree tree_options: Comma-separated tree display options: - device/nodevice, blkdevname, active, open, rw, uuid - ascii, utf, vt100, compact, inverted, notrunc exec_cmd: Execute command for each device (device name appended) Returns: If tree=True: Tree output as string Otherwise: List of device names """cmd='dmsetup ls'iftarget_type:cmd+=f' --target {target_type}'ifoutput_format!='devno':cmd+=f' -o {output_format}'iftree:cmd+=' --tree'iftree_options:cmd+=f' -o {tree_options}'ifexec_cmd:cmd+=f' --exec "{exec_cmd}"'result=run(cmd)ifresult.failed:logging.warning('No Device Mapper devices found')return[]ifnottreeelse''output=result.stdout.strip()iftreeorexec_cmd:returnoutput# Parse device names from outputdevices=[]forlineinoutput.splitlines():ifline.strip():parts=line.strip().split()ifparts:devices.append(parts[0])returndevices
defmangle(self)->bool:"""Mangle device name. Ensures the device name and UUID contains only whitelisted characters (supported by udev) and renames if necessary. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup mangle {self.dm_name}')ifresult.failed:logging.error(f'Failed to mangle {self.dm_name}')returnFalsereturnTrue
Shows the data that the device would report to the IMA subsystem
if a measurement was triggered. This is for debugging and does not
actually trigger a measurement.
defmeasure(self)->str|None:"""Show IMA measurement data. Shows the data that the device would report to the IMA subsystem if a measurement was triggered. This is for debugging and does not actually trigger a measurement. Returns: Measurement data string or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNoneresult=run(f'dmsetup measure {self.dm_name}')ifresult.failed:logging.error(f'Failed to measure {self.dm_name}')returnNonereturnresult.stdout.strip()
defmessage(self,sector:int,message:str)->bool:"""Send message to device target. Send a message to the target at the specified sector. Use sector 0 if the sector is not relevant to the target. Args: sector: Sector number (use 0 if not needed) message: Message string to send to the target Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup message {self.dm_name}{sector}{message}')ifresult.failed:logging.error(f'Failed to send message to {self.dm_name}: {result.stderr}')returnFalsereturnTrue
defmknodes(self)->bool:"""Ensure /dev/mapper node is correct. Ensures that the node in /dev/mapper for this device is correct. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup mknodes {self.dm_name}')ifresult.failed:logging.error(f'Failed to mknodes for {self.dm_name}')returnFalsereturnTrue
Ensures that all nodes in /dev/mapper correspond to mapped devices
currently loaded by the device-mapper kernel driver, adding, changing,
or removing nodes as necessary.
@classmethoddefmknodes_all(cls)->bool:"""Ensure all /dev/mapper nodes are correct. Ensures that all nodes in /dev/mapper correspond to mapped devices currently loaded by the device-mapper kernel driver, adding, changing, or removing nodes as necessary. Returns: True if successful, False otherwise """result=run('dmsetup mknodes')ifresult.failed:logging.error('Failed to run mknodes')returnFalsereturnTrue
defrefresh(self)->None:"""Refresh device state from system. Re-reads the device table and status from the system. Subclasses should override to parse target-specific attributes. Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:return# Refresh table from systemresult=run(f'dmsetup table {self.dm_name}')ifresult.succeeded:self.table=result.stdout.strip()self._parse_table()
defreload_table(self,new_table:str)->bool:"""Reload device table. Suspends the device, loads a new table, and resumes. Args: new_table: New table string to load Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalse# Suspend deviceifnotself.suspend():returnFalse# Load new tableresult=run(f'dmsetup load {self.dm_name} "{new_table}"')ifresult.failed:logging.error(f'Failed to load new table: {result.stderr}')self.resume()returnFalse# Resume with new tableifnotself.resume():returnFalse# Refresh from systemself.refresh()logging.info(f'Successfully reloaded table for {self.dm_name}')returnTrue
Removes the device mapping. The underlying devices are unaffected.
Open devices cannot be removed, but --force will replace the table with
one that fails all I/O. --deferred enables deferred removal when the device
is in use.
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
Enable deferred removal - device removed when last user closes it
False
Returns:
Type
Description
bool
True if successful, False otherwise
Note
Do NOT combine --force and --udevcookie as this may cause
nondeterministic results.
defremove(self,*,force:bool=False,retry:bool=False,deferred:bool=False,)->bool:"""Remove device. Removes the device mapping. The underlying devices are unaffected. Open devices cannot be removed, but --force will replace the table with one that fails all I/O. --deferred enables deferred removal when the device is in use. 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: Enable deferred removal - device removed when last user closes it Returns: True if successful, False otherwise Note: Do NOT combine --force and --udevcookie as this may cause nondeterministic results. """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup remove {self.dm_name}'ifforce:cmd+=' --force'ifretry:cmd+=' --retry'ifdeferred:cmd+=' --deferred'result=run(cmd)ifresult.failed:logging.error('Failed to remove device')returnFalseself.is_created=FalsereturnTrue
@classmethoddefremove_all(cls,*,force:bool=False,deferred:bool=False)->bool:"""Remove all device mapper devices. Attempts to remove all device definitions (reset the driver). This also runs mknodes afterwards. Use with care! Args: force: Replace tables with one that fails all I/O before removing deferred: Enable deferred removal for open devices Returns: True if successful, False otherwise Warning: Open devices cannot be removed unless force or deferred is used. """cmd='dmsetup remove_all'ifforce:cmd+=' --force'ifdeferred:cmd+=' --deferred'result=run(cmd)ifresult.failed:logging.error('Failed to remove all devices')returnFalsereturnTrue
defrename(self,new_name:str)->bool:"""Rename the device. Args: new_name: New name for the device Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup rename {self.dm_name}{new_name}')ifresult.failed:logging.error(f'Failed to rename {self.dm_name} to {new_name}')returnFalse# Update internal stateold_name=self.dm_nameself.dm_name=new_nameself.path=f'/dev/mapper/{new_name}'logging.info(f'Successfully renamed device from {old_name} to {new_name}')returnTrue
defresume(self,*,addnodeoncreate:bool=False,addnodeonresume:bool=False,noflush:bool=False,nolockfs:bool=False,readahead:str|int|None=None,)->bool:"""Resume device. Un-suspends a device. If an inactive table has been loaded, it becomes live. Postponed I/O then gets re-queued for processing. Args: addnodeoncreate: Ensure /dev/mapper node exists after create addnodeonresume: Ensure /dev/mapper node exists after resume (default with udev) noflush: Do not flush outstanding I/O nolockfs: Do not attempt to synchronize filesystem readahead: Read ahead size in sectors, or 'auto', 'none', or '+N' for minimum Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup resume {self.dm_name}'ifaddnodeoncreate:cmd+=' --addnodeoncreate'ifaddnodeonresume:cmd+=' --addnodeonresume'ifnoflush:cmd+=' --noflush'ifnolockfs:cmd+=' --nolockfs'ifreadaheadisnotNone:cmd+=f' --readahead {readahead}'result=run(cmd)ifresult.failed:logging.error('Failed to resume device')returnFalsereturnTrue
defset_uuid(self,uuid:str)->bool:"""Set the UUID of a device that was created without one. After a UUID has been set it cannot be changed. Args: uuid: UUID to set for the device Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup rename {self.dm_name} --setuuid {uuid}')ifresult.failed:logging.error(f'Failed to set UUID for {self.dm_name}: {result.stderr}')returnFalselogging.info(f'Successfully set UUID {uuid} for device {self.dm_name}')returnTrue
defsetgeometry(self,cylinders:int,heads:int,sectors:int,start:int,)->bool:"""Set the device geometry. Sets the device geometry to C/H/S format. Args: cylinders: Number of cylinders heads: Number of heads sectors: Sectors per track start: Start sector Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalseresult=run(f'dmsetup setgeometry {self.dm_name}{cylinders}{heads}{sectors}{start}')ifresult.failed:logging.error(f'Failed to set geometry for {self.dm_name}: {result.stderr}')returnFalsereturnTrue
Splits the device name into its subsystem components.
LVM generates device names by concatenating Volume Group, Logical Volume,
and any internal Layer with a hyphen separator.
Parameters:
Name
Type
Description
Default
subsystem
str
Subsystem to use for splitting (default: LVM)
'LVM'
Returns:
Type
Description
dict[str, str] | None
Dictionary with split name components or None on error
defsplitname(self,subsystem:str='LVM')->dict[str,str]|None:"""Split device name into subsystem constituents. Splits the device name into its subsystem components. LVM generates device names by concatenating Volume Group, Logical Volume, and any internal Layer with a hyphen separator. Args: subsystem: Subsystem to use for splitting (default: LVM) Returns: Dictionary with split name components or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNoneresult=run(f'dmsetup splitname {self.dm_name}{subsystem}')ifresult.failed:logging.error(f'Failed to split name {self.dm_name}')returnNone# Parse output into dictionaryoutput=result.stdout.strip()name_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)name_dict[key.strip()]=value.strip()returnname_dict
Suspends I/O to the device. Any I/O that has already been mapped by the device
but has not yet completed will be flushed. Any further I/O to that device will
be postponed for as long as the device is suspended.
Parameters:
Name
Type
Description
Default
nolockfs
bool
Do not attempt to synchronize filesystem (skip fsfreeze)
False
noflush
bool
Let outstanding I/O remain unflushed (requires target support)
defsuspend(self,*,nolockfs:bool=False,noflush:bool=False,)->bool:"""Suspend device. Suspends I/O to the device. Any I/O that has already been mapped by the device but has not yet completed will be flushed. Any further I/O to that device will be postponed for as long as the device is suspended. Args: nolockfs: Do not attempt to synchronize filesystem (skip fsfreeze) noflush: Let outstanding I/O remain unflushed (requires target support) Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup suspend {self.dm_name}'ifnolockfs:cmd+=' --nolockfs'ifnoflush:cmd+=' --noflush'result=run(cmd)ifresult.failed:logging.error('Failed to suspend device')returnFalsereturnTrue
@staticmethoddeftargets()->list[dict[str,str]]:"""Get list of available device mapper targets. Displays the names and versions of currently-loaded targets. Returns: List of dictionaries with 'name' and 'version' keys """result=run('dmsetup targets')ifresult.failed:logging.error('Failed to get targets')return[]targets_list:list[dict[str,str]]=[]forlineinresult.stdout.strip().splitlines():parts=line.strip().split()iflen(parts)>=2:targets_list.append({'name':parts[0],'version':parts[1],})returntargets_list
@staticmethoddefudevcomplete(cookie:str)->bool:"""Signal udev processing completion. Wakes any processes that are waiting for udev to complete processing the specified cookie. Args: cookie: Cookie value to signal completion for Returns: True if successful, False otherwise """result=run(f'dmsetup udevcomplete {cookie}')ifresult.failed:logging.error(f'Failed to complete udev cookie {cookie}')returnFalsereturnTrue
@staticmethoddefudevcomplete_all(age_in_minutes:int|None=None)->bool:"""Remove all old udev cookies. Removes all cookies older than the specified number of minutes. Any process waiting on a cookie will be resumed immediately. Args: age_in_minutes: Remove cookies older than this many minutes Returns: True if successful, False otherwise """cmd='dmsetup udevcomplete_all'ifage_in_minutesisnotNone:cmd+=f' {age_in_minutes}'result=run(cmd)ifresult.failed:logging.error('Failed to complete all udev cookies')returnFalsereturnTrue
@staticmethoddefudevcookie()->list[str]:"""List all existing udev cookies. Cookies are system-wide semaphores with keys prefixed by two predefined bytes (0x0D4D). Returns: List of cookie values """result=run('dmsetup udevcookie')ifresult.failed:logging.error('Failed to list udev cookies')return[]returnresult.stdout.strip().splitlines()
Creates a new cookie to synchronize actions with udev processing.
The cookie can be used with --udevcookie option or exported as
DM_UDEV_COOKIE environment variable.
Returns:
Type
Description
str | None
Cookie value string or None on error
Note
The cookie is a system-wide semaphore that needs to be cleaned up
explicitly by calling udevreleasecookie().
@staticmethoddefudevcreatecookie()->str|None:"""Create a new udev synchronization cookie. Creates a new cookie to synchronize actions with udev processing. The cookie can be used with --udevcookie option or exported as DM_UDEV_COOKIE environment variable. Returns: Cookie value string or None on error Note: The cookie is a system-wide semaphore that needs to be cleaned up explicitly by calling udevreleasecookie(). """result=run('dmsetup udevcreatecookie')ifresult.failed:logging.error('Failed to create udev cookie')returnNonereturnresult.stdout.strip()
@staticmethoddefudevflags(cookie:str)->dict[str,str]:"""Parse udev control flags from cookie. Parses the cookie value and extracts any udev control flags encoded. The output is in environment key format suitable for use in udev rules. Args: cookie: Cookie value to parse Returns: Dictionary of flag names to values (typically '1') """result=run(f'dmsetup udevflags {cookie}')ifresult.failed:logging.error(f'Failed to get udev flags for cookie {cookie}')return{}flags:dict[str,str]={}forlineinresult.stdout.strip().splitlines():if'='inline:key,value=line.split('=',1)flags[key.strip()]=value.strip().strip('\'"')returnflags
@staticmethoddefudevreleasecookie(cookie:str|None=None)->bool:"""Release a udev synchronization cookie. Waits for all pending udev processing bound to the cookie value and cleans up the cookie with underlying semaphore. Args: cookie: Cookie value to release. If not provided, uses DM_UDEV_COOKIE environment variable. Returns: True if successful, False otherwise """cmd='dmsetup udevreleasecookie'ifcookie:cmd+=f' {cookie}'result=run(cmd)ifresult.failed:logging.error('Failed to release udev cookie')returnFalsereturnTrue
@staticmethoddefversion()->dict[str,str]|None:"""Get dmsetup and driver version information. Returns: Dictionary with 'library' and 'driver' version strings, or None on error """result=run('dmsetup version')ifresult.failed:logging.error('Failed to get dmsetup version')returnNoneoutput=result.stdout.strip()version_dict:dict[str,str]={}forlineinoutput.splitlines():if':'inline:key,value=line.split(':',1)version_dict[key.strip().lower().replace(' ','_')]=value.strip()returnversion_dict
defwait(self,event_nr:int|None=None,*,noflush:bool=False,)->int|None:"""Wait for device event. Sleeps until the event counter for the device exceeds event_nr. Args: event_nr: Event number to wait for (waits until counter exceeds this) noflush: Do not commit thin-pool metadata before reporting Returns: Current event number after waiting, or None on error """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnNonecmd=f'dmsetup wait {self.dm_name}'ifnoflush:cmd+=' --noflush'ifevent_nrisnotNone:cmd+=f' {event_nr}'result=run(cmd)ifresult.failed:logging.error(f'Failed to wait for {self.dm_name}')returnNone# Parse event number from outputtry:returnint(result.stdout.strip())exceptValueError:returnNone
Wait for any I/O in-flight through the device to complete, then replace
the table with a new table that fails any new I/O sent to the device.
If successful, this should release any devices held open by the device's table(s).
defwipe_table(self,*,force:bool=False,noflush:bool=False,nolockfs:bool=False,)->bool:"""Wipe device table. Wait for any I/O in-flight through the device to complete, then replace the table with a new table that fails any new I/O sent to the device. If successful, this should release any devices held open by the device's table(s). Args: force: Force the operation noflush: Do not flush outstanding I/O nolockfs: Do not attempt to synchronize filesystem Returns: True if successful, False otherwise """ifnotself.is_createdornotself.dm_name:logging.error('Device not created or dm_name not available')returnFalsecmd=f'dmsetup wipe_table {self.dm_name}'ifforce:cmd+=' --force'ifnoflush:cmd+=' --noflush'ifnolockfs:cmd+=' --nolockfs'result=run(cmd)ifresult.failed:logging.error(f'Failed to wipe table for {self.dm_name}: {result.stderr}')returnFalsereturnTrue
Simulates an unreliable device that periodically fails or corrupts I/O.
Useful for testing error handling and recovery.
Args format: [ []]
Example
# Device up for 60s, down for 10s, dropping writes when downtarget=FlakeyDevice(0,1000000,'253:0 0 60 10 1 drop_writes')str(target)'0 1000000 flakey 253:0 0 60 10 1 drop_writes'
@dataclassclassFlakeyDevice(DmDevice):"""Flakey target. Simulates an unreliable device that periodically fails or corrupts I/O. Useful for testing error handling and recovery. Args format: <device> <offset> <up_interval> <down_interval> [<num_features> [<features>]] Example: ```python # Device up for 60s, down for 10s, dropping writes when down target = FlakeyDevice(0, 1000000, '253:0 0 60 10 1 drop_writes') str(target) '0 1000000 flakey 253:0 0 60 10 1 drop_writes' ``` """def__post_init__(self)->None:"""Set target type and initialize."""self.target_type='flakey'super().__post_init__()@classmethoddeffrom_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 (default: 0) size_sectors: Size in sectors (default: full device) drop_writes: Silently drop writes when down (default: False) error_writes: Error writes when down (default: False) corrupt_bio_byte: Tuple of (nth_byte, direction, value, flags) for corruption start: Start sector in virtual device (default: 0) Returns: FlakeyDevice instance Example: ```python # Device available 60s, then drops writes for 10s device = BlockDevice('/dev/sdb') flakey = FlakeyDevice.from_block_device(device, up_interval=60, down_interval=10, drop_writes=True) ``` """ifsize_sectorsisNoneanddevice.sizeisnotNone:size_sectors=device.size//device.sector_sizedevice_id=cls._get_device_identifier(device)# Build args: <device> <offset> <up_interval> <down_interval> [features]args_parts=[device_id,str(offset),str(up_interval),str(down_interval),]# Build feature listfeatures:list[str]=[]ifdrop_writes:features.append('drop_writes')iferror_writes:features.append('error_writes')ifcorrupt_bio_byteisnotNone:nth_byte,direction,value,flags=corrupt_bio_bytefeatures.append(f'corrupt_bio_byte {nth_byte}{direction}{value}{flags}')# Add features if anyiffeatures:args_parts.append(str(len(features)))args_parts.extend(features)args=' '.join(args_parts)ifsize_sectorsisNone:raiseValueError('size_sectors must be provided or device.size must be available')returncls(start=start,size_sectors=size_sectors,args=args)
# Device available 60s, then drops writes for 10sdevice=BlockDevice('/dev/sdb')flakey=FlakeyDevice.from_block_device(device,up_interval=60,down_interval=10,drop_writes=True)
@classmethoddeffrom_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 (default: 0) size_sectors: Size in sectors (default: full device) drop_writes: Silently drop writes when down (default: False) error_writes: Error writes when down (default: False) corrupt_bio_byte: Tuple of (nth_byte, direction, value, flags) for corruption start: Start sector in virtual device (default: 0) Returns: FlakeyDevice instance Example: ```python # Device available 60s, then drops writes for 10s device = BlockDevice('/dev/sdb') flakey = FlakeyDevice.from_block_device(device, up_interval=60, down_interval=10, drop_writes=True) ``` """ifsize_sectorsisNoneanddevice.sizeisnotNone:size_sectors=device.size//device.sector_sizedevice_id=cls._get_device_identifier(device)# Build args: <device> <offset> <up_interval> <down_interval> [features]args_parts=[device_id,str(offset),str(up_interval),str(down_interval),]# Build feature listfeatures:list[str]=[]ifdrop_writes:features.append('drop_writes')iferror_writes:features.append('error_writes')ifcorrupt_bio_byteisnotNone:nth_byte,direction,value,flags=corrupt_bio_bytefeatures.append(f'corrupt_bio_byte {nth_byte}{direction}{value}{flags}')# Add features if anyiffeatures:args_parts.append(str(len(features)))args_parts.extend(features)args=' '.join(args_parts)ifsize_sectorsisNone:raiseValueError('size_sectors must be provided or device.size must be available')returncls(start=start,size_sectors=size_sectors,args=args)
@dataclassclassLinearDevice(DmDevice):"""Linear device mapper device. The simplest DM device type - maps a linear range of the virtual device directly onto a linear range of another device. Args format: <destination device> <sector offset> Example: ```python device = LinearDevice.from_block_device(backing_dev, size_sectors=1000000) device.create('my-linear') str(device) '0 1000000 linear 8:16 0' ``` """def__post_init__(self)->None:"""Set target type and initialize."""self.target_type='linear'super().__post_init__()@classmethoddeffrom_block_device(cls,device:BlockDevice,start:int=0,size_sectors:int|None=None,offset:int=0,)->LinearDevice:"""Create LinearDevice from BlockDevice. Args: device: Source block device start: Start sector in virtual device (default: 0) size_sectors: Size in sectors (default: device size in sectors) offset: Offset in source device (default: 0) Returns: LinearDevice instance Example: ```python device = BlockDevice('/dev/sdb') linear = LinearDevice.from_block_device(device, size_sectors=1000000) linear.create('my-linear') ``` """ifsize_sectorsisNoneanddevice.sizeisnotNone:size_sectors=device.size//device.sector_sizedevice_id=cls._get_device_identifier(device)args=f'{device_id}{offset}'ifsize_sectorsisNone:raiseValueError('size_sectors must be provided or device.size must be available')returncls(start=start,size_sectors=size_sectors,args=args)@classmethoddefcreate_positional(cls,device_path:str,offset:int=0,start:int=0,size_sectors:int|None=None,)->LinearDevice:"""Create LinearDevice with positional arguments. Args: device_path: Target device path (e.g., '/dev/sdb' or '8:16') offset: Offset in source device sectors (default: 0) start: Start sector in virtual device (default: 0) size_sectors: Size in sectors (required) Returns: LinearDevice instance Example: ```python linear = LinearDevice.create_positional('/dev/sdb', offset=0, size_sectors=2097152) linear.create('my-linear') ``` """ifsize_sectorsisNone:raiseValueError('Size must be specified for linear devices')args=f'{device_path}{offset}'returncls(start=start,size_sectors=size_sectors,args=args)
A virtual device that allocates space from a thin pool on demand.
Enables over-provisioning of storage.
Maintains a reference to its parent pool for proper lifecycle management.
Args format:
Example
# Create via pool (recommended)thin=pool.create_thin(thin_id=0,size=2097152,dm_name='my-thin')# Or create manuallythin=ThinDevice.from_thin_pool(pool,thin_id=1,size=2097152)thin.create('another-thin')
@dataclassclassThinDevice(DmDevice):"""Thin target. A virtual device that allocates space from a thin pool on demand. Enables over-provisioning of storage. Maintains a reference to its parent pool for proper lifecycle management. Args format: <pool dev> <dev id> Example: ```python # Create via pool (recommended) thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin') # Or create manually thin = ThinDevice.from_thin_pool(pool, thin_id=1, size=2097152) thin.create('another-thin') ``` """# Reference to parent poolpool:ThinPoolDevice|None=field(init=False,default=None,repr=False)# Thin device ID within poolthin_id:int|None=field(init=False,default=None,repr=False)def__post_init__(self)->None:"""Set target type and initialize."""self.target_type='thin'super().__post_init__()@classmethoddefcreate_from_pool(cls,pool:ThinPoolDevice,thin_id:int,size:int,start:int=0,)->ThinDevice:"""Create ThinDevice configuration from pool. This is used by ThinPoolDevice.create_thin() and create_snapshot() to create the device configuration before activation. Args: pool: The parent thin pool device thin_id: Unique ID for this thin volume within the pool size: Size in sectors start: Start sector in virtual device (default: 0) Returns: ThinDevice instance (not yet activated) """pool_id=cls._get_device_identifier(pool)args=f'{pool_id}{thin_id}'device=cls(start=start,size_sectors=size,args=args)device.pool=pooldevice.thin_id=thin_idreturndevice@classmethoddeffrom_thin_pool(cls,pool_device:DmDevice,thin_id:int,start:int=0,size:int|None=None,)->ThinDevice:"""Create ThinDevice from thin pool device. Note: For full parent-child tracking, use ThinPoolDevice.create_thin() instead. Args: pool_device: The thin pool device thin_id: Unique ID for this thin volume within the pool start: Start sector in virtual device (default: 0) size: Size in sectors (required for thin targets) Returns: ThinDevice instance Raises: ValueError: If size is not provided Example: ```python pool = DmDevice(dm_name='pool') thin = ThinDevice.from_thin_pool(pool, thin_id=1, size=2097152) ``` """ifsizeisNone:raiseValueError('Size must be specified for thin targets')pool_id=cls._get_device_identifier(pool_device)args=f'{pool_id}{thin_id}'device=cls(start=start,size_sectors=size,args=args)device.thin_id=thin_id# Set pool reference if it's a ThinPoolDeviceifisinstance(pool_device,ThinPoolDevice):device.pool=pool_devicereturndevice
@classmethoddefcreate_from_pool(cls,pool:ThinPoolDevice,thin_id:int,size:int,start:int=0,)->ThinDevice:"""Create ThinDevice configuration from pool. This is used by ThinPoolDevice.create_thin() and create_snapshot() to create the device configuration before activation. Args: pool: The parent thin pool device thin_id: Unique ID for this thin volume within the pool size: Size in sectors start: Start sector in virtual device (default: 0) Returns: ThinDevice instance (not yet activated) """pool_id=cls._get_device_identifier(pool)args=f'{pool_id}{thin_id}'device=cls(start=start,size_sectors=size,args=args)device.pool=pooldevice.thin_id=thin_idreturndevice
@classmethoddeffrom_thin_pool(cls,pool_device:DmDevice,thin_id:int,start:int=0,size:int|None=None,)->ThinDevice:"""Create ThinDevice from thin pool device. Note: For full parent-child tracking, use ThinPoolDevice.create_thin() instead. Args: pool_device: The thin pool device thin_id: Unique ID for this thin volume within the pool start: Start sector in virtual device (default: 0) size: Size in sectors (required for thin targets) Returns: ThinDevice instance Raises: ValueError: If size is not provided Example: ```python pool = DmDevice(dm_name='pool') thin = ThinDevice.from_thin_pool(pool, thin_id=1, size=2097152) ``` """ifsizeisNone:raiseValueError('Size must be specified for thin targets')pool_id=cls._get_device_identifier(pool_device)args=f'{pool_id}{thin_id}'device=cls(start=start,size_sectors=size,args=args)device.thin_id=thin_id# Set pool reference if it's a ThinPoolDeviceifisinstance(pool_device,ThinPoolDevice):device.pool=pool_devicereturndevice
@dataclassclassThinPoolDevice(DmDevice):"""Thin pool target. Manages a pool of storage space from which thin volumes can be allocated. Enables thin provisioning - allocating more virtual space than physical. Tracks child ThinDevice instances and provides methods for managing them. Args format: <metadata dev> <data dev> <block size> <low water mark> <flags> <args> Example: ```python # Create pool from block devices pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev) pool.create('my-pool') # Create thin volume from pool thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin') # Create snapshot snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap') # Cleanup pool.delete_thin(1) # Delete snapshot pool.delete_thin(0) # Delete origin pool.remove() ``` """# Track child thin devices: thin_id -> ThinDevicethin_devices:dict[int,ThinDevice]=field(init=False,default_factory=dict,repr=False)def__post_init__(self)->None:"""Set target type and initialize."""self.target_type='thin-pool'super().__post_init__()@classmethoddeffrom_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. Args: metadata_device: Device for storing thin pool metadata data_device: Device for storing actual data block_size_sectors: Block size in sectors (default: 128 = 64KB) low_water_mark: Low water mark in sectors (default: 0 = auto) features: List of thin pool features (default: ['skip_block_zeroing']) start: Start sector in virtual device (default: 0) size: Size in sectors (default: data device size) Returns: ThinPoolDevice instance Example: ```python metadata_dev = BlockDevice('/dev/sdb1') # Small device for metadata data_dev = BlockDevice('/dev/sdb2') # Large device for data pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev) ``` """ifsizeisNoneanddata_device.sizeisnotNone:# Use data device size, converted to sectors using actual sector sizesize=data_device.size//data_device.sector_sizeiffeaturesisNone:features=['skip_block_zeroing']metadata_id=cls._get_device_identifier(metadata_device)data_id=cls._get_device_identifier(data_device)# 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)ifsizeisNone:raiseValueError('size must be provided or data_device.size must be available')returncls(start=start,size_sectors=size,args=args)defcreate_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. Args: thin_id: Unique ID for this thin volume within the pool size: Size in sectors for the thin volume dm_name: Device mapper name for the activated device activate: Whether to activate the device (default: True) Returns: ThinDevice instance if successful, None otherwise Example: ```python pool.create('my-pool') thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin') ``` """ifnotself.is_created:logging.error('Pool must be created before creating thin devices')returnNoneifthin_idinself.thin_devices:logging.error(f'Thin device with ID {thin_id} already exists in pool')returnNone# Send create_thin message to poolifnotself.message(0,f'"create_thin {thin_id}"'):logging.error(f'Failed to create thin device {thin_id} in pool')returnNonelogging.info(f'Created thin device ID {thin_id} in pool {self.dm_name}')# Create ThinDevice configurationthin_dev=ThinDevice.create_from_pool(pool=self,thin_id=thin_id,size=size,)ifactivateandnotthin_dev.create(dm_name):logging.error(f'Failed to activate thin device {dm_name}')# Clean up the thin device from poolself.message(0,f'"delete {thin_id}"')returnNone# Track the thin deviceself.thin_devices[thin_id]=thin_devreturnthin_devdefdelete_thin(self,thin_id:int,*,force:bool=False)->bool:"""Delete a thin volume from the pool. Removes the thin device mapping and deletes it from the pool. Args: thin_id: ID of the thin volume to delete force: Force removal even if device is in use Returns: True if successful, False otherwise Example: ```python pool.delete_thin(0) ``` """ifnotself.is_created:logging.error('Pool must be created before deleting thin devices')returnFalse# Remove the DM device if it's tracked and activatedifthin_idinself.thin_devices:thin_dev=self.thin_devices[thin_id]ifthin_dev.is_createdandnotthin_dev.remove(force=force):logging.error(f'Failed to remove thin device {thin_dev.dm_name}')returnFalse# Delete thin from poolifnotself.message(0,f'"delete {thin_id}"'):logging.error(f'Failed to delete thin device {thin_id} from pool')returnFalse# Remove from trackingself.thin_devices.pop(thin_id,None)logging.info(f'Deleted thin device ID {thin_id} from pool {self.dm_name}')returnTruedefcreate_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. Creates an internal snapshot using 'create_snap' message. The origin device should be suspended during snapshot creation. Args: origin_id: ID of the origin thin volume to snapshot snap_id: Unique ID for the new snapshot dm_name: Device mapper name for the activated snapshot size: Size in sectors (default: same as origin) activate: Whether to activate the snapshot device (default: True) Returns: ThinDevice instance for the snapshot if successful, None otherwise Example: ```python # Suspend origin before snapshot origin = pool.thin_devices[0] origin.suspend() # Create snapshot snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap') # Resume origin origin.resume() ``` """ifnotself.is_created:logging.error('Pool must be created before creating snapshots')returnNoneifsnap_idinself.thin_devices:logging.error(f'Thin device with ID {snap_id} already exists in pool')returnNone# Determine size from origin if not specifiedifsizeisNone:iforigin_idinself.thin_devices:size=self.thin_devices[origin_id].size_sectorselse:logging.error(f'Origin {origin_id} not tracked, size must be specified')returnNone# Send create_snap message to poolifnotself.message(0,f'"create_snap {snap_id}{origin_id}"'):logging.error(f'Failed to create snapshot {snap_id} of {origin_id}')returnNonelogging.info(f'Created snapshot ID {snap_id} of origin {origin_id} in pool {self.dm_name}')# Create ThinDevice configuration for snapshotsnap_dev=ThinDevice.create_from_pool(pool=self,thin_id=snap_id,size=size,)ifactivateandnotsnap_dev.create(dm_name):logging.error(f'Failed to activate snapshot device {dm_name}')# Clean up the snapshot from poolself.message(0,f'"delete {snap_id}"')returnNone# Track the snapshot deviceself.thin_devices[snap_id]=snap_devreturnsnap_devdefremove(self,*,force:bool=False,retry:bool=False,deferred:bool=False)->bool:"""Remove the thin pool device. Removes all child thin devices first, then removes the pool. Args: force: Replace table with one that fails all I/O retry: Retry removal for a few seconds if it fails deferred: Enable deferred removal when device is in use Returns: True if successful, False otherwise """# Remove all tracked thin devices firstforthin_idinlist(self.thin_devices.keys()):self.delete_thin(thin_id,force=force)# Remove the pool itselfreturnsuper().remove(force=force,retry=retry,deferred=deferred)
defcreate_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. Creates an internal snapshot using 'create_snap' message. The origin device should be suspended during snapshot creation. Args: origin_id: ID of the origin thin volume to snapshot snap_id: Unique ID for the new snapshot dm_name: Device mapper name for the activated snapshot size: Size in sectors (default: same as origin) activate: Whether to activate the snapshot device (default: True) Returns: ThinDevice instance for the snapshot if successful, None otherwise Example: ```python # Suspend origin before snapshot origin = pool.thin_devices[0] origin.suspend() # Create snapshot snap = pool.create_snapshot(origin_id=0, snap_id=1, dm_name='my-snap') # Resume origin origin.resume() ``` """ifnotself.is_created:logging.error('Pool must be created before creating snapshots')returnNoneifsnap_idinself.thin_devices:logging.error(f'Thin device with ID {snap_id} already exists in pool')returnNone# Determine size from origin if not specifiedifsizeisNone:iforigin_idinself.thin_devices:size=self.thin_devices[origin_id].size_sectorselse:logging.error(f'Origin {origin_id} not tracked, size must be specified')returnNone# Send create_snap message to poolifnotself.message(0,f'"create_snap {snap_id}{origin_id}"'):logging.error(f'Failed to create snapshot {snap_id} of {origin_id}')returnNonelogging.info(f'Created snapshot ID {snap_id} of origin {origin_id} in pool {self.dm_name}')# Create ThinDevice configuration for snapshotsnap_dev=ThinDevice.create_from_pool(pool=self,thin_id=snap_id,size=size,)ifactivateandnotsnap_dev.create(dm_name):logging.error(f'Failed to activate snapshot device {dm_name}')# Clean up the snapshot from poolself.message(0,f'"delete {snap_id}"')returnNone# Track the snapshot deviceself.thin_devices[snap_id]=snap_devreturnsnap_dev
defcreate_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. Args: thin_id: Unique ID for this thin volume within the pool size: Size in sectors for the thin volume dm_name: Device mapper name for the activated device activate: Whether to activate the device (default: True) Returns: ThinDevice instance if successful, None otherwise Example: ```python pool.create('my-pool') thin = pool.create_thin(thin_id=0, size=2097152, dm_name='my-thin') ``` """ifnotself.is_created:logging.error('Pool must be created before creating thin devices')returnNoneifthin_idinself.thin_devices:logging.error(f'Thin device with ID {thin_id} already exists in pool')returnNone# Send create_thin message to poolifnotself.message(0,f'"create_thin {thin_id}"'):logging.error(f'Failed to create thin device {thin_id} in pool')returnNonelogging.info(f'Created thin device ID {thin_id} in pool {self.dm_name}')# Create ThinDevice configurationthin_dev=ThinDevice.create_from_pool(pool=self,thin_id=thin_id,size=size,)ifactivateandnotthin_dev.create(dm_name):logging.error(f'Failed to activate thin device {dm_name}')# Clean up the thin device from poolself.message(0,f'"delete {thin_id}"')returnNone# Track the thin deviceself.thin_devices[thin_id]=thin_devreturnthin_dev
defdelete_thin(self,thin_id:int,*,force:bool=False)->bool:"""Delete a thin volume from the pool. Removes the thin device mapping and deletes it from the pool. Args: thin_id: ID of the thin volume to delete force: Force removal even if device is in use Returns: True if successful, False otherwise Example: ```python pool.delete_thin(0) ``` """ifnotself.is_created:logging.error('Pool must be created before deleting thin devices')returnFalse# Remove the DM device if it's tracked and activatedifthin_idinself.thin_devices:thin_dev=self.thin_devices[thin_id]ifthin_dev.is_createdandnotthin_dev.remove(force=force):logging.error(f'Failed to remove thin device {thin_dev.dm_name}')returnFalse# Delete thin from poolifnotself.message(0,f'"delete {thin_id}"'):logging.error(f'Failed to delete thin device {thin_id} from pool')returnFalse# Remove from trackingself.thin_devices.pop(thin_id,None)logging.info(f'Deleted thin device ID {thin_id} from pool {self.dm_name}')returnTrue
metadata_dev=BlockDevice('/dev/sdb1')# Small device for metadatadata_dev=BlockDevice('/dev/sdb2')# Large device for datapool=ThinPoolDevice.from_block_devices(metadata_dev,data_dev)
@classmethoddeffrom_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. Args: metadata_device: Device for storing thin pool metadata data_device: Device for storing actual data block_size_sectors: Block size in sectors (default: 128 = 64KB) low_water_mark: Low water mark in sectors (default: 0 = auto) features: List of thin pool features (default: ['skip_block_zeroing']) start: Start sector in virtual device (default: 0) size: Size in sectors (default: data device size) Returns: ThinPoolDevice instance Example: ```python metadata_dev = BlockDevice('/dev/sdb1') # Small device for metadata data_dev = BlockDevice('/dev/sdb2') # Large device for data pool = ThinPoolDevice.from_block_devices(metadata_dev, data_dev) ``` """ifsizeisNoneanddata_device.sizeisnotNone:# Use data device size, converted to sectors using actual sector sizesize=data_device.size//data_device.sector_sizeiffeaturesisNone:features=['skip_block_zeroing']metadata_id=cls._get_device_identifier(metadata_device)data_id=cls._get_device_identifier(data_device)# 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)ifsizeisNone:raiseValueError('size must be provided or data_device.size must be available')returncls(start=start,size_sectors=size,args=args)
defremove(self,*,force:bool=False,retry:bool=False,deferred:bool=False)->bool:"""Remove the thin pool device. Removes all child thin devices first, then removes the pool. Args: force: Replace table with one that fails all I/O retry: Retry removal for a few seconds if it fails deferred: Enable deferred removal when device is in use Returns: True if successful, False otherwise """# Remove all tracked thin devices firstforthin_idinlist(self.thin_devices.keys()):self.delete_thin(thin_id,force=force)# Remove the pool itselfreturnsuper().remove(force=force,retry=retry,deferred=deferred)
Provides block-level deduplication, compression, and thin provisioning.
VDO is a device mapper target that can add these features to the storage stack.
Note: VDO volumes must be formatted with 'vdoformat' before use.
Storage block device (must be pre-formatted with vdoformat)
required
logical_size_sectors
int
Logical device size in 512-byte sectors
required
start
int
Start sector in virtual device (default: 0)
0
minimum_io_size
int
Minimum I/O size in bytes, 512 or 4096 (default: 4096)
4096
block_map_cache_size_mb
int
Block map cache size in MB (default: 128, min: 128)
128
block_map_period
int
Block map era length (default: 16380, range: 1-16380)
16380
**kwargs
int | bool | None
Optional VDO parameters (use dmsetup parameter names):
ack: Number of ack threads (default: 1)
bio: Number of bio threads (default: 4)
bioRotationInterval: Bio rotation interval (default: 64, range: 1-1024)
cpu: Number of CPU threads (default: 1)
hash: Number of hash zone threads (default: 0, range: 0-100)
logical: Number of logical threads (default: 0, range: 0-60)
physical: Number of physical threads (default: 0, range: 0-16)
maxDiscard: Maximum discard size in 4096-byte blocks (default: 1)
deduplication: Enable deduplication (default: True)
compression: Enable compression (default: False)
@classmethoddeffrom_block_device(cls,device:BlockDevice,logical_size_sectors:int,*,start:int=0,minimum_io_size:int=4096,block_map_cache_size_mb:int=128,block_map_period:int=16380,**kwargs:int|bool|None,)->VdoDevice:"""Create VdoTarget from BlockDevice. Args: device: Storage block device (must be pre-formatted with vdoformat) logical_size_sectors: Logical device size in 512-byte sectors start: Start sector in virtual device (default: 0) minimum_io_size: Minimum I/O size in bytes, 512 or 4096 (default: 4096) block_map_cache_size_mb: Block map cache size in MB (default: 128, min: 128) block_map_period: Block map era length (default: 16380, range: 1-16380) **kwargs: Optional VDO parameters (use dmsetup parameter names): ack: Number of ack threads (default: 1) bio: Number of bio threads (default: 4) bioRotationInterval: Bio rotation interval (default: 64, range: 1-1024) cpu: Number of CPU threads (default: 1) hash: Number of hash zone threads (default: 0, range: 0-100) logical: Number of logical threads (default: 0, range: 0-60) physical: Number of physical threads (default: 0, range: 0-16) maxDiscard: Maximum discard size in 4096-byte blocks (default: 1) deduplication: Enable deduplication (default: True) compression: Enable compression (default: False) Returns: VdoTarget instance Example: ```python device = BlockDevice('/dev/sdb') # Pre-formatted with vdoformat target = VdoTarget.from_block_device( device, logical_size_sectors=2097152, # 1 GB logical compression=True, ) ``` """# Get device identifierdevice_id=cls._get_device_identifier(device)# Calculate storage device size in 4096-byte blocksifdevice.sizeisNone:raiseValueError('Cannot determine size: device size is unknown')storage_size_blocks=device.size//4096# Convert block map cache size from MB to 4096-byte blocksblock_map_cache_blocks=(block_map_cache_size_mb*1024*1024)//4096# Build required argumentsargs_parts=['V4',device_id,str(storage_size_blocks),str(minimum_io_size),str(block_map_cache_blocks),str(block_map_period),]# Build optional arguments as key-value pairs# NOTE: dm-vdo kernel constraint: if any of hash/logical/physical threads is# non-zero, all three must be specified and non-zero. This is NOT validated here# to allow testing negative cases - the kernel will reject invalid configurations.# Valid dmsetup parameter names for thread/discard settingsoptional_params=('ack','bio','bioRotationInterval','cpu','hash','logical','physical','maxDiscard')optional_args:list[str]=[]forparam_nameinoptional_params:value=kwargs.get(param_name)ifvalueisnotNone:optional_args.extend([param_name,str(value)])# Handle deduplication (only add if False, as True is default)dedup_val=kwargs.get('deduplication',True)deduplication_enabled=bool(dedup_val)ifdedup_valisnotNoneelseTrueifnotdeduplication_enabled:optional_args.extend(['deduplication','off'])# Handle compression (only add if True, as False is default)comp_val=kwargs.get('compression',False)compression_enabled=bool(comp_val)ifcomp_valisnotNoneelseFalseifcompression_enabled:optional_args.extend(['compression','on'])# Combine all argumentsifoptional_args:args_parts.extend(optional_args)args=' '.join(args_parts)target=cls(start=start,size_sectors=logical_size_sectors,args=args)# Set attributes directly - we know the valuestarget.vdo_version='V4'target.storage_device=device_idtarget.storage_size_blocks=storage_size_blockstarget.minimum_io_size=minimum_io_sizetarget.block_map_cache_blocks=block_map_cache_blockstarget.block_map_period=block_map_periodtarget.deduplication=deduplication_enabledtarget.compression=compression_enabled# Set optional parameters from kwargsforparam_nameinoptional_params:value=kwargs.get(param_name)ifvalueisnotNone:setattr(target,param_name,value)returntarget
@classmethoddeffrom_table_line(cls,table_line:str)->VdoDevice|None:"""Create VdoTarget from a dmsetup table line. Parses a full table line (including start, size, and type) and creates a VdoTarget instance. Useful for reconstructing targets from existing devices. Args: table_line: Full table line from 'dmsetup table' output Returns: VdoTarget instance or None if parsing fails Example: ```python line = '0 2097152 vdo V4 8:16 262144 4096 32768 16380' target = VdoTarget.from_table_line(line) target.size # 2097152 ``` """parts=table_line.strip().split(None,3)iflen(parts)<4:logging.warning(f'Invalid table line: {table_line}')returnNonestart=int(parts[0])size=int(parts[1])target_type=parts[2]args=parts[3]iftarget_type!='vdo':logging.warning(f'Not a VDO target: {target_type}')returnNonetarget=cls(start=start,size_sectors=size,args=args)target.refresh()# Parse args to populate attributesreturntarget
@staticmethoddefparse_status(status_line:str)->dict[str,str|int]:"""Parse VDO status output into a dictionary. Parses the output from 'dmsetup status' for a VDO device. Status format from dmsetup: <start> <size> vdo <device> <operating mode> <in recovery> <index state> <compression state> <physical blocks used> <total physical blocks> Args: status_line: Status line from 'dmsetup status' output Returns: Dictionary with parsed status fields: - start: Start sector - size: Size in sectors - device: VDO storage device - operating_mode: 'normal', 'recovering', or 'read-only' - in_recovery: 'recovering' or '-' - index_state: 'closed', 'closing', 'error', 'offline', 'online', 'opening', 'unknown' - compression_state: 'offline' or 'online' - physical_blocks_used: Number of physical blocks in use - total_physical_blocks: Total physical blocks available Example: ```python status = '0 2097152 vdo /dev/loop0 normal - online online 12345 262144' parsed = VdoDevice.parse_status(status) parsed['operating_mode'] # 'normal' parsed['physical_blocks_used'] # 12345 ``` """parts=status_line.strip().split()result:dict[str,str|int]={}# Format: <start> <size> vdo <device> <mode> <recovery> <index> <compression> <used> <total># Minimum 10 parts expectediflen(parts)<10:logging.warning(f'Invalid VDO status line (expected 10+ parts, got {len(parts)}): {status_line}')returnresulttry:result['start']=int(parts[0])result['size']=int(parts[1])exceptValueError:logging.warning(f'Failed to parse start/size from status: {status_line}')returnresult# parts[2] is 'vdo' (target type)result['device']=parts[3]result['operating_mode']=parts[4]result['in_recovery']=parts[5]result['index_state']=parts[6]result['compression_state']=parts[7]try:result['physical_blocks_used']=int(parts[8])result['total_physical_blocks']=int(parts[9])exceptValueError:logging.warning(f'Failed to parse block counts from status: {status_line}')returnresult
Extracts all VDO parameters from the args string and updates
instance attributes. Useful after modifying args or when
reconstructing from dmsetup table output.
Device Mapper Persistent Data (DMPD) tools management.
This module provides functionality for managing Device Mapper Persistent Data tools.
Each tool is a separate command-line utility for managing metadata of device-mapper targets:
Era Tools:
- era_check: Check era metadata integrity
- era_dump: Dump era metadata to file or stdout
- era_repair: Repair corrupted era metadata
- era_restore: Restore era metadata from backup
Each tool operates independently and can work with various device sources:
- Block devices (/dev/sdX, /dev/mapper/*, /dev/vg/lv)
- Regular files (metadata dumps)
- Device mapper devices
classCacheDump(DmpdTool):"""Dump cache metadata to stdout or file. Usage: cache_dump [options] {device|file} For available options, see: cache_dump --help """TOOL_NAME='cache_dump'
Each DMPD tool is a separate command-line utility that can operate on:
- Block devices (LVM logical volumes, raw devices)
- Regular files (metadata dump files)
- Device mapper devices
classDmpdTool(ABC):"""Base class for DMPD command-line tools. Each DMPD tool is a separate command-line utility that can operate on: - Block devices (LVM logical volumes, raw devices) - Regular files (metadata dump files) - Device mapper devices """# Tool name (must be set by subclasses)TOOL_NAME:ClassVar[str]@staticmethoddef_build_command(cmd:str,device_or_file:str|Path|None=None,**options:CommandOption)->str:"""Build DMPD command string with options. Args: cmd: Command name (e.g., 'cache_check') device_or_file: Device path or file path **options: Command options Returns: Complete command string """command_parts=[cmd]# Add device/file path if providedifdevice_or_file:command_parts.append(str(device_or_file))# Add optionsforoption_key,valueinoptions.items():ifvalueisnotNone:key=option_key.replace('_','-')ifisinstance(value,bool)andvalue:command_parts.append(f'--{key}')else:command_parts.append(f'--{key}={value!s}')return' '.join(command_parts)@classmethoddefrun(cls,device_or_file:str|Path|None=None,**options:CommandOption)->CommandResult:"""Run the DMPD tool with specified options. Args: device_or_file: Device path or file path to operate on **options: Tool-specific options Returns: Command result """cmd=cls._build_command(cls.TOOL_NAME,device_or_file,**options)returnrun(cmd)@classmethoddefhelp(cls)->CommandResult:"""Get help information for the tool."""returnrun(f'{cls.TOOL_NAME} --help')@classmethoddefversion(cls)->CommandResult:"""Get version information for the tool."""returnrun(f'{cls.TOOL_NAME} --version')
@classmethoddefrun(cls,device_or_file:str|Path|None=None,**options:CommandOption)->CommandResult:"""Run the DMPD tool with specified options. Args: device_or_file: Device path or file path to operate on **options: Tool-specific options Returns: Command result """cmd=cls._build_command(cls.TOOL_NAME,device_or_file,**options)returnrun(cmd)
classEraCheck(DmpdTool):"""Check era metadata integrity. Usage: era_check [options] {device|file} For available options, see: era_check --help """TOOL_NAME='era_check'
classEraDump(DmpdTool):"""Dump era metadata to stdout or file. Usage: era_dump [options] {device|file} For available options, see: era_dump --help """TOOL_NAME='era_dump'
classEraRestore(DmpdTool):"""Restore era metadata from backup. Usage: era_restore [options] --input <file> --output {device|file} For available options, see: era_restore --help """TOOL_NAME='era_restore'
classThinDump(DmpdTool):"""Dump thin metadata to stdout or file. Usage: thin_dump [options] {device|file} For available options, see: thin_dump --help """TOOL_NAME='thin_dump'
defera_check(device_or_file:str|Path,**options:CommandOption)->CommandResult:"""Check era metadata integrity."""returnEraCheck.run(device_or_file,**options)
defthin_delta(device_or_file:str|Path,**options:CommandOption)->CommandResult:"""Print differences between thin devices."""returnThinDelta.run(device_or_file,**options)