Skip to content

Multipath

DM-Multipath aggregates multiple I/O paths to a storage device (e.g., dual FC or iSCSI links) into a single /dev/mapper/mpath* device, providing path failover and load balancing. Managed by the multipathd daemon.

sts.multipath

Multipath device management.

Device discovery, path/service management, and configuration for DM multipath.

MultipathDevice pydantic-model

Bases: BlockDevice

Multipath device managed by multipathd.

For creating DM multipath targets directly (without multipathd), see sts.dm.multipath.MultipathTarget.

The dm attribute provides access to a DmDevice for low-level DM operations (table, size). size is derived from the DM device.

Example 
device = MultipathDevice().discover()  # Uses first available device
device = MultipathDevice(name='mpatha').discover()
device.dm.table  # Access DM table
Show JSON schema: 
{
  "$defs": {
    "BlockdevInfo": {
      "description": "Parsed ``blockdev --report`` output.",
      "properties": {
        "ro": {
          "default": false,
          "title": "Ro",
          "type": "boolean"
        },
        "ra": {
          "default": 0,
          "title": "Ra",
          "type": "integer"
        },
        "log-sec": {
          "default": 0,
          "title": "Log-Sec",
          "type": "integer"
        },
        "phy-sec": {
          "default": 0,
          "title": "Phy-Sec",
          "type": "integer"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size": {
          "default": 0,
          "title": "Size",
          "type": "integer"
        }
      },
      "title": "BlockdevInfo",
      "type": "object"
    },
    "DmDevice": {
      "additionalProperties": false,
      "description": "Base class for all Device Mapper devices.\n\nBefore ``create()``, the device is just a target configuration\n(start, size_sectors, args). After ``create()``, it becomes a full\nblock device with dm_name, path, table, and all BlockDevice functionality.",
      "properties": {
        "path": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "format": "path",
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Path"
        },
        "name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Name"
        },
        "size": {
          "anyOf": [
            {
              "minimum": 0,
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Size"
        },
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "blockdev_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/BlockdevInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "lsblk_info": {
          "anyOf": [
            {
              "$ref": "#/$defs/LsblkInfo"
            },
            {
              "type": "null"
            }
          ],
          "default": null
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        },
        "size_sectors": {
          "default": 0,
          "title": "Size Sectors",
          "type": "integer"
        },
        "args": {
          "default": "",
          "title": "Args",
          "type": "string"
        },
        "dm_name": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Dm Name"
        },
        "target_type": {
          "default": "",
          "title": "Target Type",
          "type": "string"
        },
        "table": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Table"
        },
        "is_created": {
          "default": false,
          "title": "Is Created",
          "type": "boolean"
        }
      },
      "title": "DmDevice",
      "type": "object"
    },
    "LsblkInfo": {
      "description": "Parsed ``lsblk -JOb`` output.",
      "properties": {
        "model": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Model"
        },
        "rm": {
          "default": false,
          "title": "Rm",
          "type": "boolean"
        },
        "hctl": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Hctl"
        },
        "state": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "State"
        },
        "pttype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pttype"
        },
        "wwn": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Wwn"
        },
        "fstype": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Fstype"
        },
        "mountpoint": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Mountpoint"
        },
        "type": {
          "default": "disk",
          "title": "Type",
          "type": "string"
        },
        "tran": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Tran"
        },
        "maj:min": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Maj:Min"
        },
        "pkname": {
          "anyOf": [
            {
              "type": "string"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Pkname"
        },
        "start": {
          "default": 0,
          "title": "Start",
          "type": "integer"
        }
      },
      "title": "LsblkInfo",
      "type": "object"
    },
    "PathGroup": {
      "description": "A path group within a multipath device's 'path_groups' report.\n\nThe inner `paths` entries keep their raw dict shape (`chk_st`, `dev`,\n`dm_st`, etc.) since the exact keys reported by multipathd vary by version.",
      "properties": {
        "group": {
          "default": 0,
          "title": "Group",
          "type": "integer"
        },
        "status": {
          "default": "",
          "title": "Status",
          "type": "string"
        },
        "paths": {
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "title": "Paths",
          "type": "array"
        }
      },
      "title": "PathGroup",
      "type": "object"
    }
  },
  "additionalProperties": false,
  "description": "Multipath device managed by multipathd.\n\nFor creating DM multipath targets directly (without multipathd), see\n`sts.dm.multipath.MultipathTarget`.\n\nThe ``dm`` attribute provides access to a `DmDevice` for low-level\nDM operations (table, size). ``size`` is derived from the DM device.\n\nExample:\n    ```python\n    device = MultipathDevice().discover()  # Uses first available device\n    device = MultipathDevice(name='mpatha').discover()\n    device.dm.table  # Access DM table\n    ```",
  "properties": {
    "path": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "format": "path",
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Path"
    },
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Name"
    },
    "size": {
      "anyOf": [
        {
          "minimum": 0,
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size"
    },
    "model": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Model"
    },
    "blockdev_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/BlockdevInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "lsblk_info": {
      "anyOf": [
        {
          "$ref": "#/$defs/LsblkInfo"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    },
    "dm_name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm Name"
    },
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "wwid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Wwid"
    },
    "vendor": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Vendor"
    },
    "n_paths": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "N Paths"
    },
    "size_str": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size Str"
    },
    "features": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Features"
    },
    "hwhandler": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Hwhandler"
    },
    "failback": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Failback"
    },
    "dm_st": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm St"
    },
    "path_groups": {
      "items": {
        "$ref": "#/$defs/PathGroup"
      },
      "title": "Path Groups",
      "type": "array"
    },
    "dm": {
      "anyOf": [
        {
          "$ref": "#/$defs/DmDevice"
        },
        {
          "type": "null"
        }
      ],
      "default": null
    }
  },
  "title": "MultipathDevice",
  "type": "object"
}

Fields:

  • path (PathOrStr | None)
  • name (str | None)
  • size (int | None)
  • model (str | None)
  • blockdev_info (BlockdevInfo | None)
  • lsblk_info (LsblkInfo | None)
  • dm_name (str | None)
  • uuid (str | None)
  • wwid (str | None)
  • vendor (str | None)
  • n_paths (int | None)
  • size_str (str | None)
  • features (str | None)
  • hwhandler (str | None)
  • failback (str | None)
  • dm_st (str | None)
  • path_groups (list[PathGroup])
  • dm (DmDevice | None)

Validators:

  • _derive_fields
Source code in sts_libs/src/sts/multipath.py 
 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
class MultipathDevice(BlockDevice):
    """Multipath device managed by multipathd.

    For creating DM multipath targets directly (without multipathd), see
    `sts.dm.multipath.MultipathTarget`.

    The ``dm`` attribute provides access to a `DmDevice` for low-level
    DM operations (table, size). ``size`` is derived from the DM device.

    Example:
        ```python
        device = MultipathDevice().discover()  # Uses first available device
        device = MultipathDevice(name='mpatha').discover()
        device.dm.table  # Access DM table
        ```
    """

    # Optional parameters - note: path, name, size, model come from BlockDevice
    dm_name: 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
    n_paths: int | None = None  # Number of paths

    # Additional optional parameters from multipathd
    size_str: str | None = None  # Size as a string (e.g. '10.0G')
    features: str | None = None  # Features of the device
    hwhandler: str | None = None  # Hardware handler
    failback: str | None = None  # Failback policy (e.g. 'immediate', 'manual', or seconds)
    dm_st: str | None = None  # Device mapper state (e.g. 'active', 'suspended')
    path_groups: list[PathGroup] = Field(default_factory=list)  # Path groups

    # DmDevice for DM-specific operations (initialized in _derive_fields)
    dm: DmDevice | None = Field(default=None, init=False, repr=False)

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

    @model_validator(mode='after')
    def _derive_fields(self) -> Self:
        """Derive path/name fields only — no I/O.

        Call discover() after construction to load state for existing devices.
        """
        if not self.path and self.name:
            self.path = Path(f'/dev/mapper/{self.name}')
        elif self.path and not self.name:
            self.name = Path(self.path).name
        return self

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

        Finds first available device if name not provided, queries
        multipathd for map data, and initializes a DmDevice for
        low-level DM operations.
        """
        # 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 = Path(f'/dev/mapper/{self.name}')

        super().discover()

        # Get device information if name provided
        if self.name:
            result = run(f'multipathd show map {self.name} json')
            if result.succeeded:
                map_data = json.loads(result.stdout).get('map', {})
                report = MultipathMapReport.model_validate(map_data)
                self._apply_map_report(report)

        # Initialize DmDevice for DM-specific operations
        self._init_dm_device()

        # Override size from DM device (size_sectors * 512)
        if self.dm and self.dm.size_sectors:
            self.size = self.dm.size_sectors * 512

        return self

    def _init_dm_device(self) -> None:
        """Initialize DmDevice for low-level DM operations."""
        if self.dm_name or self.path:
            try:
                self.dm = DmDevice(dm_name=self.dm_name, path=self.path).discover()
            except DeviceError:
                logger.debug(f'Could not initialize DmDevice for {self.name}')
                self.dm = None

    @property
    def table(self) -> str | None:
        """Device Mapper table string, or None if unavailable."""
        if self.dm:
            return self.dm.table
        return None

    def _apply_map_report(self, report: MultipathMapReport) -> None:
        """Apply parsed map report data to device attributes."""
        if report.uuid:
            self.uuid = report.uuid
            self.wwid = report.uuid
        if report.sysfs:
            self.dm_name = report.sysfs
        if report.failback:
            self.failback = report.failback
        if report.prod:
            self.model = report.prod
        if report.vend:
            self.vendor = report.vend
        if report.paths is not None:
            self.n_paths = report.paths
        if report.dm_st:
            self.dm_st = report.dm_st
        if report.size_str:
            self.size_str = report.size_str
        if report.features:
            self.features = report.features
        if report.hwhandler:
            self.hwhandler = report.hwhandler
        if report.path_groups:
            self.path_groups = report.path_groups

    def suspend(self) -> bool:
        """Suspend the multipath device (keeps paths configured)."""
        if not self.name:
            logger.error('Device name not available')
            return False

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

    def resume(self) -> CommandResult:
        """Resume a suspended multipath device.

        Raises:
            DeviceError: If device name is not available.
        """
        if not self.name:
            raise DeviceError('Device name not available')

        return run(f'multipath -a {self.name}')

    def remove(self) -> CommandResult:
        """Remove the multipath device (flushes I/O and clears DM table).

        Raises:
            DeviceError: If device name is not available.
        """
        if not self.name:
            raise DeviceError('Device name not available')

        return run(f'multipath -f {self.name}')

    @property
    def paths(self) -> list[dict[str, Any]]:
        """All path detail dicts flattened from path groups."""
        return [path for group in self.path_groups for path in group.paths]

    @classmethod
    def get_all(cls) -> list[MultipathDevice]:
        """Discover all multipath devices on the system."""
        result = run('multipath -ll -v1')
        if result.failed:
            logger.warning(f'Running "multipath -ll -v1" failed:\n{result.stderr}')
            return []

        return [cls(name=dev).discover() for dev in result.stdout.splitlines() if dev.strip()]

    @classmethod
    def get_by_wwid(cls, wwid: str) -> MultipathDevice | None:
        """Find a multipath device by its WWID."""
        if not wwid:
            msg = 'WWID required'
            raise ValueError(msg)

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

        return None

    @classmethod
    def get_by_vendor(cls, vendor: str) -> list[MultipathDevice]:
        """Find multipath devices matching a vendor string."""
        if not vendor:
            msg = 'Vendor required'
            raise ValueError(msg)

        devices = cls.get_all()
        return [device for device in devices if device.vendor == vendor]

paths property

All path detail dicts flattened from path groups.

table property

Device Mapper table string, or None if unavailable.

discover()

Load multipath device state from the system.

Finds first available device if name not provided, queries multipathd for map data, and initializes a DmDevice for low-level DM operations.

Source code in sts_libs/src/sts/multipath.py 
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
def discover(self) -> Self:
    """Load multipath device state from the system.

    Finds first available device if name not provided, queries
    multipathd for map data, and initializes a DmDevice for
    low-level DM operations.
    """
    # 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 = Path(f'/dev/mapper/{self.name}')

    super().discover()

    # Get device information if name provided
    if self.name:
        result = run(f'multipathd show map {self.name} json')
        if result.succeeded:
            map_data = json.loads(result.stdout).get('map', {})
            report = MultipathMapReport.model_validate(map_data)
            self._apply_map_report(report)

    # Initialize DmDevice for DM-specific operations
    self._init_dm_device()

    # Override size from DM device (size_sectors * 512)
    if self.dm and self.dm.size_sectors:
        self.size = self.dm.size_sectors * 512

    return self

get_all() classmethod

Discover all multipath devices on the system.

Source code in sts_libs/src/sts/multipath.py 
231
232
233
234
235
236
237
238
239
@classmethod
def get_all(cls) -> list[MultipathDevice]:
    """Discover all multipath devices on the system."""
    result = run('multipath -ll -v1')
    if result.failed:
        logger.warning(f'Running "multipath -ll -v1" failed:\n{result.stderr}')
        return []

    return [cls(name=dev).discover() for dev in result.stdout.splitlines() if dev.strip()]

get_by_vendor(vendor) classmethod

Find multipath devices matching a vendor string.

Source code in sts_libs/src/sts/multipath.py 
254
255
256
257
258
259
260
261
262
@classmethod
def get_by_vendor(cls, vendor: str) -> list[MultipathDevice]:
    """Find multipath devices matching a vendor string."""
    if not vendor:
        msg = 'Vendor required'
        raise ValueError(msg)

    devices = cls.get_all()
    return [device for device in devices if device.vendor == vendor]

get_by_wwid(wwid) classmethod

Find a multipath device by its WWID.

Source code in sts_libs/src/sts/multipath.py 
241
242
243
244
245
246
247
248
249
250
251
252
@classmethod
def get_by_wwid(cls, wwid: str) -> MultipathDevice | None:
    """Find a multipath device by its WWID."""
    if not wwid:
        msg = 'WWID required'
        raise ValueError(msg)

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

    return None

remove()

Remove the multipath device (flushes I/O and clears DM table).

Raises:

Type Description
DeviceError

If device name is not available.
Source code in sts_libs/src/sts/multipath.py 
215
216
217
218
219
220
221
222
223
224
def remove(self) -> CommandResult:
    """Remove the multipath device (flushes I/O and clears DM table).

    Raises:
        DeviceError: If device name is not available.
    """
    if not self.name:
        raise DeviceError('Device name not available')

    return run(f'multipath -f {self.name}')

resume()

Resume a suspended multipath device.

Raises:

Type Description
DeviceError

If device name is not available.
Source code in sts_libs/src/sts/multipath.py 
204
205
206
207
208
209
210
211
212
213
def resume(self) -> CommandResult:
    """Resume a suspended multipath device.

    Raises:
        DeviceError: If device name is not available.
    """
    if not self.name:
        raise DeviceError('Device name not available')

    return run(f'multipath -a {self.name}')

suspend()

Suspend the multipath device (keeps paths configured).

Source code in sts_libs/src/sts/multipath.py 
192
193
194
195
196
197
198
199
200
201
202
def suspend(self) -> bool:
    """Suspend the multipath device (keeps paths configured)."""
    if not self.name:
        logger.error('Device name not available')
        return False

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

MultipathMapReport pydantic-model

Bases: ReportModel

Parsed output of 'multipathd show map json'.

Show JSON schema: 
{
  "$defs": {
    "PathGroup": {
      "description": "A path group within a multipath device's 'path_groups' report.\n\nThe inner `paths` entries keep their raw dict shape (`chk_st`, `dev`,\n`dm_st`, etc.) since the exact keys reported by multipathd vary by version.",
      "properties": {
        "group": {
          "default": 0,
          "title": "Group",
          "type": "integer"
        },
        "status": {
          "default": "",
          "title": "Status",
          "type": "string"
        },
        "paths": {
          "items": {
            "additionalProperties": true,
            "type": "object"
          },
          "title": "Paths",
          "type": "array"
        }
      },
      "title": "PathGroup",
      "type": "object"
    }
  },
  "description": "Parsed output of 'multipathd show map <name> json'.",
  "properties": {
    "uuid": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Uuid"
    },
    "sysfs": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Sysfs"
    },
    "failback": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Failback"
    },
    "prod": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Prod"
    },
    "vend": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Vend"
    },
    "paths": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Paths"
    },
    "dm_st": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Dm St"
    },
    "size_str": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Size Str"
    },
    "features": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Features"
    },
    "hwhandler": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Hwhandler"
    },
    "path_groups": {
      "items": {
        "$ref": "#/$defs/PathGroup"
      },
      "title": "Path Groups",
      "type": "array"
    }
  },
  "title": "MultipathMapReport",
  "type": "object"
}

Fields:

  • uuid (str | None)
  • sysfs (str | None)
  • failback (str | None)
  • prod (str | None)
  • vend (str | None)
  • paths (int | None)
  • dm_st (str | None)
  • size_str (str | None)
  • features (str | None)
  • hwhandler (str | None)
  • path_groups (list[PathGroup])
Source code in sts_libs/src/sts/multipath.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class MultipathMapReport(ReportModel):
    """Parsed output of 'multipathd show map <name> json'."""

    uuid: str | None = None
    sysfs: str | None = None
    failback: str | None = None
    prod: str | None = None
    vend: str | None = None
    paths: int | None = None
    dm_st: str | None = None
    size_str: str | None = None
    features: str | None = None
    hwhandler: str | None = None
    path_groups: list[PathGroup] = Field(default_factory=list)

MultipathService

Stateless wrapper for the multipathd systemd service.

Plain Python class (not Pydantic) -- no fields to validate.

Source code in sts_libs/src/sts/multipath.py 
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
331
332
333
334
335
336
337
338
339
340
class MultipathService:
    """Stateless wrapper for the multipathd systemd service.

    Plain Python class (not Pydantic) -- no fields to validate.
    """

    def __init__(self) -> None:
        self.config_path = MultipathDevice.MULTIPATH_CONF

    def start(self) -> bool:
        """Start multipathd, creating default config if needed."""
        # Create default config if needed
        if not self.config_path.exists():
            result = run('mpathconf --enable')
            if result.failed:
                logger.error('Failed to create default config')
                return False

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

        return True

    @staticmethod
    def stop() -> bool:
        """Stop multipathd."""
        result = run('systemctl stop multipathd')
        if result.failed:
            logger.error('Failed to stop multipathd')
            return False

        return True

    @staticmethod
    def reload() -> bool:
        """Reload multipathd configuration without restart."""
        result = run('systemctl reload multipathd')
        if result.failed:
            logger.error('Failed to reload multipathd')
            return False

        return True

    @staticmethod
    def is_running() -> bool:
        """Check if multipathd is running."""
        result = run('systemctl is-active multipathd')
        return result.succeeded

    @staticmethod
    def configure(
        find_multipaths: Literal['yes', 'no', 'strict', 'greedy', 'smart'] | None = None,
    ) -> bool:
        """Run ``mpathconf --enable`` with optional ``find_multipaths`` mode."""
        cmd = ['mpathconf', '--enable']
        if find_multipaths:
            cmd.extend(['--find_multipaths', find_multipaths])

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

        return True

    @staticmethod
    def flush() -> bool:
        """Flush all unused multipath devices."""
        result = run('multipath -F')
        if result.failed:
            logger.error('Failed to flush devices')
            return False

        return True

configure(find_multipaths=None) staticmethod

Run mpathconf --enable with optional find_multipaths mode.

Source code in sts_libs/src/sts/multipath.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
@staticmethod
def configure(
    find_multipaths: Literal['yes', 'no', 'strict', 'greedy', 'smart'] | None = None,
) -> bool:
    """Run ``mpathconf --enable`` with optional ``find_multipaths`` mode."""
    cmd = ['mpathconf', '--enable']
    if find_multipaths:
        cmd.extend(['--find_multipaths', find_multipaths])

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

    return True

flush() staticmethod

Flush all unused multipath devices.

Source code in sts_libs/src/sts/multipath.py 
332
333
334
335
336
337
338
339
340
@staticmethod
def flush() -> bool:
    """Flush all unused multipath devices."""
    result = run('multipath -F')
    if result.failed:
        logger.error('Failed to flush devices')
        return False

    return True

is_running() staticmethod

Check if multipathd is running.

Source code in sts_libs/src/sts/multipath.py 
310
311
312
313
314
@staticmethod
def is_running() -> bool:
    """Check if multipathd is running."""
    result = run('systemctl is-active multipathd')
    return result.succeeded

reload() staticmethod

Reload multipathd configuration without restart.

Source code in sts_libs/src/sts/multipath.py 
300
301
302
303
304
305
306
307
308
@staticmethod
def reload() -> bool:
    """Reload multipathd configuration without restart."""
    result = run('systemctl reload multipathd')
    if result.failed:
        logger.error('Failed to reload multipathd')
        return False

    return True

start()

Start multipathd, creating default config if needed.

Source code in sts_libs/src/sts/multipath.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def start(self) -> bool:
    """Start multipathd, creating default config if needed."""
    # Create default config if needed
    if not self.config_path.exists():
        result = run('mpathconf --enable')
        if result.failed:
            logger.error('Failed to create default config')
            return False

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

    return True

stop() staticmethod

Stop multipathd.

Source code in sts_libs/src/sts/multipath.py 
290
291
292
293
294
295
296
297
298
@staticmethod
def stop() -> bool:
    """Stop multipathd."""
    result = run('systemctl stop multipathd')
    if result.failed:
        logger.error('Failed to stop multipathd')
        return False

    return True

PathGroup pydantic-model

Bases: ReportModel

A path group within a multipath device's 'path_groups' report.

The inner paths entries keep their raw dict shape (chk_st, dev, dm_st, etc.) since the exact keys reported by multipathd vary by version.

Show JSON schema: 
{
  "description": "A path group within a multipath device's 'path_groups' report.\n\nThe inner `paths` entries keep their raw dict shape (`chk_st`, `dev`,\n`dm_st`, etc.) since the exact keys reported by multipathd vary by version.",
  "properties": {
    "group": {
      "default": 0,
      "title": "Group",
      "type": "integer"
    },
    "status": {
      "default": "",
      "title": "Status",
      "type": "string"
    },
    "paths": {
      "items": {
        "additionalProperties": true,
        "type": "object"
      },
      "title": "Paths",
      "type": "array"
    }
  },
  "title": "PathGroup",
  "type": "object"
}

Config:

  • frozen: True
  • extra: ignore

Fields:

  • group (int)
  • status (str)
  • paths (list[dict[str, Any]])
Source code in sts_libs/src/sts/multipath.py 
33
34
35
36
37
38
39
40
41
42
class PathGroup(ReportModel):
    """A path group within a multipath device's 'path_groups' report.

    The inner `paths` entries keep their raw dict shape (`chk_st`, `dev`,
    `dm_st`, etc.) since the exact keys reported by multipathd vary by version.
    """

    group: int = 0
    status: str = ''
    paths: list[dict[str, Any]] = Field(default_factory=list)