Skip to content

Multipath

This section documents the Device Mapper Multipath functionality.

sts.multipath

Multipath device management.

This module provides functionality for managing multipath devices: - Device discovery - Path management - Configuration management - Service management

Device Mapper Multipath provides: - I/O failover for redundancy - I/O load balancing for performance - Automatic path management - Consistent device naming

Common use cases: - High availability storage - SAN connectivity - iSCSI with multiple NICs - FC with multiple HBAs

MultipathDevice dataclass

Bases: DmDevice

Multipath device representation.

A multipath device combines multiple physical paths to the same storage into a single virtual device. This provides: - Automatic failover if paths fail - Load balancing across paths - Consistent device naming

Parameters:

Name Type Description Default
name str | None

Device name (optional, defaults to first available mpathX)

 None
path Path | str | None

Device path (optional, defaults to /dev/mapper/)

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
dm_name str | None

Device Mapper name (optional, discovered from device)

 None
model str | None

Device model (optional)

 None
uuid str | None

Device UUID (optional)

 None
wwid str | None

Device WWID (optional, discovered from device)

 None
vendor str | None

Device vendor (optional)

 None
paths list[dict[str, Any]]

List of paths to device (HCTL and device node)

 list()
Example 
print('kokot')
device = MultipathDevice()  # Uses first available device
device = MultipathDevice(name='mpatha')  # Uses specific device
Source code in sts_libs/src/sts/multipath.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
@dataclass
class MultipathDevice(DmDevice):
    """Multipath device representation.

    A multipath device combines multiple physical paths to the same storage
    into a single virtual device. This provides:
    - Automatic failover if paths fail
    - Load balancing across paths
    - Consistent device naming

    Args:
        name: Device name (optional, defaults to first available mpathX)
        path: Device path (optional, defaults to /dev/mapper/<name>)
        size: Device size in bytes (optional, discovered from device)
        dm_name: Device Mapper name (optional, discovered from device)
        model: Device model (optional)
        uuid: Device UUID (optional)
        wwid: Device WWID (optional, discovered from device)
        vendor: Device vendor (optional)
        paths: List of paths to device (HCTL and device node)

    Example:
        ```python
        print('kokot')
        device = MultipathDevice()  # Uses first available device
        device = MultipathDevice(name='mpatha')  # Uses specific device
        ```
    """

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

    # Optional parameters for this class
    wwid: str | None = None  # World Wide ID (unique identifier)
    vendor: str | None = None  # Device vendor
    paths: list[dict[str, Any]] = field(default_factory=list)  # Available paths

    # Configuration file paths
    MULTIPATH_CONF: ClassVar[Path] = Path('/etc/multipath.conf')
    MULTIPATH_BINDINGS: ClassVar[Path] = Path('/etc/multipath/bindings')

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

        - Finds first available device if name not provided
        - Sets device path if not provided
        - Discovers device information and paths

        Raises:
            DeviceNotFoundError: If device does not exist
            DeviceError: If device cannot be accessed
        """
        # Get first available device if name not provided
        if not self.name:
            result = run('multipath -ll -v1')
            if result.succeeded and result.stdout:
                self.name = result.stdout.split()[0]

        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/mapper/{self.name}'

        # Initialize parent class
        super().__post_init__()

        # Get device information if name provided
        if self.name:
            result = run(f'multipath -ll {self.name}')
            if result.succeeded:
                self._parse_device_info(result.stdout)

    def _parse_device_info(self, output: str) -> None:
        """Parse multipath -ll output for device information.

        Extracts key device information:
        - Device mapper name
        - World Wide ID (WWID)
        - Vendor and model
        - Path information

        Args:
            output: Output from multipath -ll command
        """
        # Parse first line for device info
        # Format: mpatha (360000000000000000e00000000000001) dm-0 VENDOR,PRODUCT
        if match := re.match(r'(\S+)\s+\((\S+)\)\s+(\S+)\s+([^,]+),(.+)', output.splitlines()[0]):
            if not self.dm_name:
                self.dm_name = match.group(3)
            if not self.wwid:
                self.wwid = match.group(2)
            if not self.vendor:
                self.vendor = match.group(4)
            if not self.model:
                self.model = match.group(5)

        # Parse paths section
        self._parse_paths(output)

    def _parse_paths(self, output: str) -> None:
        """Parse multipath -ll output for path information.

        Extracts information about each path:
        - SCSI HCTL (Host:Channel:Target:LUN)
        - Device node (e.g. sda)
        - Path state and priority

        Args:
            output: Output from multipath -ll command
        """
        current_path: dict[str, Any] = {}
        for line in output.splitlines():
            # Skip policy lines (separate path groups)
            if 'policy' in line:
                if current_path:
                    self.paths.append(current_path)
                    current_path = {}
                continue

            # Parse path line
            # Format: 1:0:0:0 sda 8:0   active ready running
            if match := re.match(r'\s*(\d+:\d+:\d+:\d+)\s+(\w+)\s+', line):
                if current_path:
                    self.paths.append(current_path)
                current_path = {
                    'hctl': match.group(1),  # SCSI address
                    'dev': match.group(2),  # Device node
                }

        # Add last path if any
        if current_path:
            self.paths.append(current_path)

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

        Temporarily disables the multipath device:
        - Stops using device for I/O
        - Keeps paths configured
        - Device can be resumed later

        Returns:
            True if successful, False otherwise

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

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

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

        Re-enables a suspended multipath device:
        - Rescans paths
        - Restores device operation
        - Resumes I/O handling

        Returns:
            True if successful, False otherwise

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

        result = run(f'multipath -a {self.name}')
        if result.failed:
            logging.error('Failed to resume device')
            return False
        return True

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

        Completely removes the multipath device:
        - Flushes I/O
        - Removes device mapper table
        - Clears path information

        Returns:
            True if successful, False otherwise

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

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

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

        Lists all configured multipath devices:
        - Includes active and inactive devices
        - Provides basic device information
        - Does not include detailed path status

        Returns:
            List of MultipathDevice instances

        Example:
            ```python
            MultipathDevice.get_all()
            [MultipathDevice(name='mpatha', ...), MultipathDevice(name='mpathb', ...)]
            ```
        """
        devices = []
        result = run('multipath -ll -v2')
        if result.failed:
            logging.warning('No multipath devices found')
            return []

        # Parse line like: mpatha (360000000000000000e00000000000001) dm-0 VENDOR,PRODUCT
        # or 360a98000324669436c2b45666c567863 dm-6 VENDOR,PRODUCT
        patterns = [r'(\S+)\s+(\S+)\s+([^,]+),(.+)', r'(\S+)\s+\((\S+)\)\s+(\S+)\s+([^,]+),(.+)']
        for line in result.stdout.splitlines():
            try:
                match = next((m for p in patterns if (m := re.match(p, line))), None)
                if match:
                    groups = match.groups()
                    wwid, dm_name, vendor, product = groups[:4]
                    name = groups[0] if len(groups) == 5 else None
                    devices.append(cls(name=name, dm_name=dm_name, wwid=wwid, vendor=vendor, model=product))
            except (ValueError, DeviceError):  # noqa: PERF203
                logging.exception(f'Failed to parse device info for line: {line}')

        return devices

    @classmethod
    def get_by_wwid(cls, wwid: str) -> MultipathDevice | None:
        """Get multipath device by WWID.

        The World Wide ID uniquely identifies a storage device:
        - Consistent across reboots
        - Same for all paths to device
        - Vendor-specific format

        Args:
            wwid: Device WWID

        Returns:
            MultipathDevice instance or None if not found

        Example:
            ```python
            MultipathDevice.get_by_wwid('360000000000000000e00000000000001')
            MultipathDevice(name='mpatha', ...)
            ```
        """
        if not wwid:
            msg = 'WWID required'
            raise ValueError(msg)

        for device in cls.get_all():
            if isinstance(device, MultipathDevice) and device.wwid == wwid:
                return device

        return None

__post_init__()

Initialize multipath device.

  • Finds first available device if name not provided
  • Sets device path if not provided
  • Discovers device information and paths

Raises:

Type Description
DeviceNotFoundError

If device does not exist
DeviceError

If device cannot be accessed
Source code in sts_libs/src/sts/multipath.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def __post_init__(self) -> None:
    """Initialize multipath device.

    - Finds first available device if name not provided
    - Sets device path if not provided
    - Discovers device information and paths

    Raises:
        DeviceNotFoundError: If device does not exist
        DeviceError: If device cannot be accessed
    """
    # Get first available device if name not provided
    if not self.name:
        result = run('multipath -ll -v1')
        if result.succeeded and result.stdout:
            self.name = result.stdout.split()[0]

    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/mapper/{self.name}'

    # Initialize parent class
    super().__post_init__()

    # Get device information if name provided
    if self.name:
        result = run(f'multipath -ll {self.name}')
        if result.succeeded:
            self._parse_device_info(result.stdout)

get_all() classmethod

Get list of all multipath devices.

Lists all configured multipath devices: - Includes active and inactive devices - Provides basic device information - Does not include detailed path status

Returns:

Type Description
Sequence[DmDevice]

List of MultipathDevice instances
Example 
MultipathDevice.get_all()
[MultipathDevice(name='mpatha', ...), MultipathDevice(name='mpathb', ...)]
Source code in sts_libs/src/sts/multipath.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
@classmethod
def get_all(cls) -> Sequence[DmDevice]:
    """Get list of all multipath devices.

    Lists all configured multipath devices:
    - Includes active and inactive devices
    - Provides basic device information
    - Does not include detailed path status

    Returns:
        List of MultipathDevice instances

    Example:
        ```python
        MultipathDevice.get_all()
        [MultipathDevice(name='mpatha', ...), MultipathDevice(name='mpathb', ...)]
        ```
    """
    devices = []
    result = run('multipath -ll -v2')
    if result.failed:
        logging.warning('No multipath devices found')
        return []

    # Parse line like: mpatha (360000000000000000e00000000000001) dm-0 VENDOR,PRODUCT
    # or 360a98000324669436c2b45666c567863 dm-6 VENDOR,PRODUCT
    patterns = [r'(\S+)\s+(\S+)\s+([^,]+),(.+)', r'(\S+)\s+\((\S+)\)\s+(\S+)\s+([^,]+),(.+)']
    for line in result.stdout.splitlines():
        try:
            match = next((m for p in patterns if (m := re.match(p, line))), None)
            if match:
                groups = match.groups()
                wwid, dm_name, vendor, product = groups[:4]
                name = groups[0] if len(groups) == 5 else None
                devices.append(cls(name=name, dm_name=dm_name, wwid=wwid, vendor=vendor, model=product))
        except (ValueError, DeviceError):  # noqa: PERF203
            logging.exception(f'Failed to parse device info for line: {line}')

    return devices

get_by_wwid(wwid) classmethod

Get multipath device by WWID.

The World Wide ID uniquely identifies a storage device: - Consistent across reboots - Same for all paths to device - Vendor-specific format

Parameters:

Name Type Description Default
wwid str

Device WWID

 required

Returns:

Type Description
MultipathDevice | None

MultipathDevice instance or None if not found
Example 
MultipathDevice.get_by_wwid('360000000000000000e00000000000001')
MultipathDevice(name='mpatha', ...)
Source code in sts_libs/src/sts/multipath.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
@classmethod
def get_by_wwid(cls, wwid: str) -> MultipathDevice | None:
    """Get multipath device by WWID.

    The World Wide ID uniquely identifies a storage device:
    - Consistent across reboots
    - Same for all paths to device
    - Vendor-specific format

    Args:
        wwid: Device WWID

    Returns:
        MultipathDevice instance or None if not found

    Example:
        ```python
        MultipathDevice.get_by_wwid('360000000000000000e00000000000001')
        MultipathDevice(name='mpatha', ...)
        ```
    """
    if not wwid:
        msg = 'WWID required'
        raise ValueError(msg)

    for device in cls.get_all():
        if isinstance(device, MultipathDevice) and device.wwid == wwid:
            return device

    return None

remove()

Remove device.

Completely removes the multipath device: - Flushes I/O - Removes device mapper table - Clears path information

Returns:

Type Description
bool

True if successful, False otherwise
Example 
device.remove()
True
Source code in sts_libs/src/sts/multipath.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def remove(self) -> bool:
    """Remove device.

    Completely removes the multipath device:
    - Flushes I/O
    - Removes device mapper table
    - Clears path information

    Returns:
        True if successful, False otherwise

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

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

resume()

Resume device.

Re-enables a suspended multipath device: - Rescans paths - Restores device operation - Resumes I/O handling

Returns:

Type Description
bool

True if successful, False otherwise
Example 
device.resume()
True
Source code in sts_libs/src/sts/multipath.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
def resume(self) -> bool:
    """Resume device.

    Re-enables a suspended multipath device:
    - Rescans paths
    - Restores device operation
    - Resumes I/O handling

    Returns:
        True if successful, False otherwise

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

    result = run(f'multipath -a {self.name}')
    if result.failed:
        logging.error('Failed to resume device')
        return False
    return True

suspend()

Suspend device.

Temporarily disables the multipath device: - Stops using device for I/O - Keeps paths configured - Device can be resumed later

Returns:

Type Description
bool

True if successful, False otherwise
Example 
device.suspend()
True
Source code in sts_libs/src/sts/multipath.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def suspend(self) -> bool:
    """Suspend device.

    Temporarily disables the multipath device:
    - Stops using device for I/O
    - Keeps paths configured
    - Device can be resumed later

    Returns:
        True if successful, False otherwise

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

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

MultipathService

Multipath service management.

Manages the multipathd service which: - Monitors path status - Handles path failures - Manages device creation - Applies configuration

Example 
service = MultipathService()
service.start()
True
Source code in sts_libs/src/sts/multipath.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
class MultipathService:
    """Multipath service management.

    Manages the multipathd service which:
    - Monitors path status
    - Handles path failures
    - Manages device creation
    - Applies configuration

    Example:
        ```python
        service = MultipathService()
        service.start()
        True
        ```
    """

    def __init__(self) -> None:
        """Initialize multipath service."""
        self.config_path = MultipathDevice.MULTIPATH_CONF
        # Ensure package is installed
        system = SystemManager()
        if not system.package_manager.install(PACKAGE_NAME):
            logging.critical(f'Could not install {PACKAGE_NAME}')

    def start(self) -> bool:
        """Start multipath service.

        Starts the multipathd daemon:
        - Creates default config if needed
        - Starts systemd service
        - Begins path monitoring

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            service.start()
            True
            ```
        """
        # Create default config if needed
        if not self.config_path.exists():
            result = run('mpathconf --enable')
            if result.failed:
                logging.error('Failed to create default config')
                return False

        result = run('systemctl start multipathd')
        if result.failed:
            logging.error('Failed to start multipathd')
            return False

        return True

    def stop(self) -> bool:
        """Stop multipath service.

        Stops the multipathd daemon:
        - Stops path monitoring
        - Keeps devices configured
        - Maintains configuration

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            service.stop()
            True
            ```
        """
        result = run('systemctl stop multipathd')
        if result.failed:
            logging.error('Failed to stop multipathd')
            return False

        return True

    def reload(self) -> bool:
        """Reload multipath configuration.

        Reloads configuration without restart:
        - Applies config changes
        - Keeps devices active
        - Updates path settings

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            service.reload()
            True
            ```
        """
        result = run('systemctl reload multipathd')
        if result.failed:
            logging.error('Failed to reload multipathd')
            return False

        return True

    def is_running(self) -> bool:
        """Check if multipath service is running.

        Returns:
            True if running, False otherwise

        Example:
            ```python
            service.is_running()
            True
            ```
        """
        result = run('systemctl is-active multipathd')
        return result.succeeded

    def configure(
        self,
        find_multipaths: Literal['yes', 'no', 'strict', 'greedy', 'smart'] | None = None,
    ) -> bool:
        """Configure multipath service.

        Sets up multipath configuration:
        - find_multipaths modes:
          - yes: Create multipath devices for likely candidates
          - no: Only create explicitly configured devices
          - strict: Only create devices with multiple paths
          - greedy: Create devices for all SCSI devices
          - smart: Create devices based on WWID patterns

        Args:
            find_multipaths: How to detect multipath devices

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            service.configure(find_multipaths='yes')
            True
            ```
        """
        cmd = ['mpathconf', '--enable']
        if find_multipaths:
            cmd.extend(['--find_multipaths', find_multipaths])

        result = run(' '.join(cmd))
        if result.failed:
            logging.error('Failed to configure multipathd')
            return False

        return True

    def get_paths(self, device: str) -> list[dict[str, Any]]:
        """Get paths for device.

        Lists all paths for a device with status:
        - Device node (e.g. sda)
        - DM state (active/passive)
        - Path state (ready/failed)
        - Online state (running/offline)

        Args:
            device: Device name or WWID

        Returns:
            List of path dictionaries

        Example:
            ```python
            service.get_paths('mpatha')
            [{'hctl': '1:0:0:0', 'dev': 'sda', 'state': 'active'}, ...]
            ```
        """
        result = run('multipathd show paths format "%d %t %T"')
        if result.failed:
            return []

        paths = []
        for line in result.stdout.splitlines():
            if device not in line:
                continue

            # Parse line like: sda active ready running
            parts = line.split()
            if len(parts) < 4:
                continue

            paths.append(
                {
                    'dev': parts[0],  # Device node
                    'dm_state': parts[1],  # Device mapper state
                    'path_state': parts[2],  # Path health
                    'online_state': parts[3],  # Connection status
                }
            )

        return paths

    def flush(self) -> bool:
        """Flush all unused multipath devices.

        Removes unused multipath devices:
        - Clears device mapper tables
        - Removes path groups
        - Keeps configuration intact

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            service.flush()
            True
            ```
        """
        result = run('multipath -F')
        if result.failed:
            logging.error('Failed to flush devices')
            return False

        return True

__init__()

Initialize multipath service.

Source code in sts_libs/src/sts/multipath.py 
350
351
352
353
354
355
356
def __init__(self) -> None:
    """Initialize multipath service."""
    self.config_path = MultipathDevice.MULTIPATH_CONF
    # Ensure package is installed
    system = SystemManager()
    if not system.package_manager.install(PACKAGE_NAME):
        logging.critical(f'Could not install {PACKAGE_NAME}')

configure(find_multipaths=None)

Configure multipath service.

Sets up multipath configuration: - find_multipaths modes: - yes: Create multipath devices for likely candidates - no: Only create explicitly configured devices - strict: Only create devices with multiple paths - greedy: Create devices for all SCSI devices - smart: Create devices based on WWID patterns

Parameters:

Name Type Description Default
find_multipaths Literal['yes', 'no', 'strict', 'greedy', 'smart'] | None

How to detect multipath devices

 None

Returns:

Type Description
bool

True if successful, False otherwise
Example 
service.configure(find_multipaths='yes')
True
Source code in sts_libs/src/sts/multipath.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def configure(
    self,
    find_multipaths: Literal['yes', 'no', 'strict', 'greedy', 'smart'] | None = None,
) -> bool:
    """Configure multipath service.

    Sets up multipath configuration:
    - find_multipaths modes:
      - yes: Create multipath devices for likely candidates
      - no: Only create explicitly configured devices
      - strict: Only create devices with multiple paths
      - greedy: Create devices for all SCSI devices
      - smart: Create devices based on WWID patterns

    Args:
        find_multipaths: How to detect multipath devices

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        service.configure(find_multipaths='yes')
        True
        ```
    """
    cmd = ['mpathconf', '--enable']
    if find_multipaths:
        cmd.extend(['--find_multipaths', find_multipaths])

    result = run(' '.join(cmd))
    if result.failed:
        logging.error('Failed to configure multipathd')
        return False

    return True

flush()

Flush all unused multipath devices.

Removes unused multipath devices: - Clears device mapper tables - Removes path groups - Keeps configuration intact

Returns:

Type Description
bool

True if successful, False otherwise
Example 
service.flush()
True
Source code in sts_libs/src/sts/multipath.py 
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
def flush(self) -> bool:
    """Flush all unused multipath devices.

    Removes unused multipath devices:
    - Clears device mapper tables
    - Removes path groups
    - Keeps configuration intact

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        service.flush()
        True
        ```
    """
    result = run('multipath -F')
    if result.failed:
        logging.error('Failed to flush devices')
        return False

    return True

get_paths(device)

Get paths for device.

Lists all paths for a device with status: - Device node (e.g. sda) - DM state (active/passive) - Path state (ready/failed) - Online state (running/offline)

Parameters:

Name Type Description Default
device str

Device name or WWID

 required

Returns:

Type Description
list[dict[str, Any]]

List of path dictionaries
Example 
service.get_paths('mpatha')
[{'hctl': '1:0:0:0', 'dev': 'sda', 'state': 'active'}, ...]
Source code in sts_libs/src/sts/multipath.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def get_paths(self, device: str) -> list[dict[str, Any]]:
    """Get paths for device.

    Lists all paths for a device with status:
    - Device node (e.g. sda)
    - DM state (active/passive)
    - Path state (ready/failed)
    - Online state (running/offline)

    Args:
        device: Device name or WWID

    Returns:
        List of path dictionaries

    Example:
        ```python
        service.get_paths('mpatha')
        [{'hctl': '1:0:0:0', 'dev': 'sda', 'state': 'active'}, ...]
        ```
    """
    result = run('multipathd show paths format "%d %t %T"')
    if result.failed:
        return []

    paths = []
    for line in result.stdout.splitlines():
        if device not in line:
            continue

        # Parse line like: sda active ready running
        parts = line.split()
        if len(parts) < 4:
            continue

        paths.append(
            {
                'dev': parts[0],  # Device node
                'dm_state': parts[1],  # Device mapper state
                'path_state': parts[2],  # Path health
                'online_state': parts[3],  # Connection status
            }
        )

    return paths

is_running()

Check if multipath service is running.

Returns:

Type Description
bool

True if running, False otherwise
Example 
service.is_running()
True
Source code in sts_libs/src/sts/multipath.py 
437
438
439
440
441
442
443
444
445
446
447
448
449
450
def is_running(self) -> bool:
    """Check if multipath service is running.

    Returns:
        True if running, False otherwise

    Example:
        ```python
        service.is_running()
        True
        ```
    """
    result = run('systemctl is-active multipathd')
    return result.succeeded

reload()

Reload multipath configuration.

Reloads configuration without restart: - Applies config changes - Keeps devices active - Updates path settings

Returns:

Type Description
bool

True if successful, False otherwise
Example 
service.reload()
True
Source code in sts_libs/src/sts/multipath.py 
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
def reload(self) -> bool:
    """Reload multipath configuration.

    Reloads configuration without restart:
    - Applies config changes
    - Keeps devices active
    - Updates path settings

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        service.reload()
        True
        ```
    """
    result = run('systemctl reload multipathd')
    if result.failed:
        logging.error('Failed to reload multipathd')
        return False

    return True

start()

Start multipath service.

Starts the multipathd daemon: - Creates default config if needed - Starts systemd service - Begins path monitoring

Returns:

Type Description
bool

True if successful, False otherwise
Example 
service.start()
True
Source code in sts_libs/src/sts/multipath.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
def start(self) -> bool:
    """Start multipath service.

    Starts the multipathd daemon:
    - Creates default config if needed
    - Starts systemd service
    - Begins path monitoring

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        service.start()
        True
        ```
    """
    # Create default config if needed
    if not self.config_path.exists():
        result = run('mpathconf --enable')
        if result.failed:
            logging.error('Failed to create default config')
            return False

    result = run('systemctl start multipathd')
    if result.failed:
        logging.error('Failed to start multipathd')
        return False

    return True

stop()

Stop multipath service.

Stops the multipathd daemon: - Stops path monitoring - Keeps devices configured - Maintains configuration

Returns:

Type Description
bool

True if successful, False otherwise
Example 
service.stop()
True
Source code in sts_libs/src/sts/multipath.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def stop(self) -> bool:
    """Stop multipath service.

    Stops the multipathd daemon:
    - Stops path monitoring
    - Keeps devices configured
    - Maintains configuration

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        service.stop()
        True
        ```
    """
    result = run('systemctl stop multipathd')
    if result.failed:
        logging.error('Failed to stop multipathd')
        return False

    return True