Skip to content

LVM

This section documents the Logical Volume Management (LVM) functionality.

sts.lvm

LVM device management.

This module provides functionality for managing LVM devices: - Physical Volume (PV) operations - Volume Group (VG) operations - Logical Volume (LV) operations

LVM (Logical Volume Management) provides flexible disk space management: 1. Physical Volumes (PVs): Physical disks or partitions 2. Volume Groups (VGs): Pool of space from PVs 3. Logical Volumes (LVs): Virtual partitions from VG space

Key benefits: - Resize filesystems online - Snapshot and mirror volumes - Stripe across multiple disks - Move data between disks

LVReport dataclass

Logical Volume report data.

This class provides detailed information about a Logical Volume from 'lvs -o lv_all --reportformat json'. Contains all available LV attributes that can be queried.

Parameters:

Name Type Description Default
name str | None

LV name (optional, used for fetching, can be discovered)

 None
vg str | None

Volume group name (optional, used for fetching, can be discovered)

 None
prevent_update bool

Flag to prevent updates from report (defaults to False)

 False
All lv_* fields

LV attributes from lvs output

 required
raw_data dict[str, Any]

Complete raw data from lvs output

 dict()
Source code in sts_libs/src/sts/lvm.py 
 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
331
332
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
@dataclass
class LVReport:
    """Logical Volume report data.

    This class provides detailed information about a Logical Volume from 'lvs -o lv_all --reportformat json'.
    Contains all available LV attributes that can be queried.

    Args:
        name: LV name (optional, used for fetching, can be discovered)
        vg: Volume group name (optional, used for fetching, can be discovered)
        prevent_update: Flag to prevent updates from report (defaults to False)

        All lv_* fields: LV attributes from lvs output
        raw_data: Complete raw data from lvs output
    """

    # Control fields
    name: str | None = None
    vg: str | None = None
    prevent_update: bool = field(default=False)

    # Core LV identification
    lv_uuid: str | None = None
    lv_name: str | None = None
    lv_full_name: str | None = None
    lv_path: str | None = None
    lv_dm_path: str | None = None
    vg_name: str | None = None

    # Size and layout information
    lv_size: str | None = None
    lv_metadata_size: str | None = None
    seg_count: str | None = None
    lv_layout: str | None = None
    lv_role: str | None = None

    # Status and attributes
    lv_attr: str | None = None
    lv_active: str | None = None
    lv_active_locally: str | None = None
    lv_active_remotely: str | None = None
    lv_active_exclusively: str | None = None
    lv_permissions: str | None = None
    lv_suspended: str | None = None

    # Device information
    lv_major: str | None = None
    lv_minor: str | None = None
    lv_kernel_major: str | None = None
    lv_kernel_minor: str | None = None
    lv_read_ahead: str | None = None
    lv_kernel_read_ahead: str | None = None

    # Pool and thin provisioning
    pool_lv: str | None = None
    pool_lv_uuid: str | None = None
    data_lv: str | None = None
    data_lv_uuid: str | None = None
    metadata_lv: str | None = None
    metadata_lv_uuid: str | None = None
    data_percent: str | None = None
    metadata_percent: str | None = None

    # Snapshot information
    origin: str | None = None
    origin_uuid: str | None = None
    origin_size: str | None = None
    snap_percent: str | None = None

    # RAID information
    raid_mismatch_count: str | None = None
    raid_sync_action: str | None = None
    raid_write_behind: str | None = None
    raid_min_recovery_rate: str | None = None
    raid_max_recovery_rate: str | None = None

    # Cache information
    cache_total_blocks: str | None = None
    cache_used_blocks: str | None = None
    cache_dirty_blocks: str | None = None
    cache_read_hits: str | None = None
    cache_read_misses: str | None = None
    cache_write_hits: str | None = None
    cache_write_misses: str | None = None
    kernel_cache_settings: str | None = None
    kernel_cache_policy: str | None = None

    # VDO information
    vdo_operating_mode: str | None = None
    vdo_compression_state: str | None = None
    vdo_index_state: str | None = None
    vdo_used_size: str | None = None
    vdo_saving_percent: str | None = None

    # Write cache information
    writecache_block_size: str | None = None
    writecache_total_blocks: str | None = None
    writecache_free_blocks: str | None = None
    writecache_writeback_blocks: str | None = None
    writecache_error: str | None = None

    # Configuration and policy
    lv_allocation_policy: str | None = None
    lv_allocation_locked: str | None = None
    lv_autoactivation: str | None = None
    lv_when_full: str | None = None
    lv_skip_activation: str | None = None
    lv_fixed_minor: str | None = None

    # Timing and host information
    lv_time: str | None = None
    lv_time_removed: str | None = None
    lv_host: str | None = None

    # Health and status checks
    lv_health_status: str | None = None
    lv_check_needed: str | None = None
    lv_merge_failed: str | None = None
    lv_snapshot_invalid: str | None = None

    # Miscellaneous
    lv_tags: str | None = None
    lv_profile: str | None = None
    lv_lockargs: str | None = None
    lv_modules: str | None = None
    lv_historical: str | None = None
    kernel_discards: str | None = None
    copy_percent: str | None = None
    sync_percent: str | None = None

    # Device table status
    lv_live_table: str | None = None
    lv_inactive_table: str | None = None
    lv_device_open: str | None = None

    # Hierarchical relationships
    lv_parent: str | None = None
    lv_ancestors: str | None = None
    lv_full_ancestors: str | None = None
    lv_descendants: str | None = None
    lv_full_descendants: str | None = None

    # Conversion and movement
    lv_converting: str | None = None
    lv_merging: str | None = None
    move_pv: str | None = None
    move_pv_uuid: str | None = None
    convert_lv: str | None = None
    convert_lv_uuid: str | None = None

    # Mirror information
    mirror_log: str | None = None
    mirror_log_uuid: str | None = None

    # Synchronization
    lv_initial_image_sync: str | None = None
    lv_image_synced: str | None = None

    # Integrity
    raidintegritymode: str | None = None
    raidintegrityblocksize: str | None = None
    integritymismatches: str | None = None
    kernel_metadata_format: str | None = None

    # Segment information (from seg_all)
    segtype: str | None = None
    stripes: str | None = None
    data_stripes: str | None = None
    stripe_size: str | None = None
    region_size: str | None = None
    chunk_size: str | None = None
    seg_start: str | None = None
    seg_start_pe: str | None = None
    seg_size: str | None = None
    seg_size_pe: str | None = None
    seg_tags: str | None = None
    seg_pe_ranges: str | None = None
    seg_le_ranges: str | None = None
    seg_metadata_le_ranges: str | None = None
    devices: str | None = None
    metadata_devices: str | None = None
    seg_monitor: str | None = None

    # Additional segment fields
    reshape_len: str | None = None
    reshape_len_le: str | None = None
    data_copies: str | None = None
    data_offset: str | None = None
    new_data_offset: str | None = None
    parity_chunks: str | None = None
    thin_count: str | None = None
    discards: str | None = None
    cache_metadata_format: str | None = None
    cache_mode: str | None = None
    zero: str | None = None
    transaction_id: str | None = None
    thin_id: str | None = None
    cache_policy: str | None = None
    cache_settings: str | None = None
    integrity_settings: str | None = None

    # VDO segment settings
    vdo_compression: str | None = None
    vdo_deduplication: str | None = None
    vdo_minimum_io_size: str | None = None
    vdo_block_map_cache_size: str | None = None
    vdo_block_map_era_length: str | None = None
    vdo_use_sparse_index: str | None = None
    vdo_index_memory_size: str | None = None
    vdo_slab_size: str | None = None
    vdo_ack_threads: str | None = None
    vdo_bio_threads: str | None = None
    vdo_bio_rotation: str | None = None
    vdo_cpu_threads: str | None = None
    vdo_hash_zone_threads: str | None = None
    vdo_logical_threads: str | None = None
    vdo_physical_threads: str | None = None
    vdo_max_discard: str | None = None
    vdo_header_size: str | None = None
    vdo_use_metadata_hints: str | None = None
    vdo_write_policy: str | None = None

    # Raw data storage for any additional fields
    raw_data: dict[str, Any] = field(default_factory=dict, repr=False)

    def __post_init__(self) -> None:
        """Initialize the report."""
        # If name and vg are provided, fetch the report data
        if self.name and self.vg:
            self.refresh()

    def refresh(self) -> bool:
        """Refresh LV report data from system.

        Updates all fields with the latest information from lvs command.

        Returns:
            bool: True if refresh was successful, False otherwise
        """
        # If prevent_update is True, skip refresh
        if self.prevent_update:
            logging.debug('Refresh skipped due to prevent_update flag')
            return True

        if not self.name or not self.vg:
            logging.error('LV name and VG name required for refresh')
            return False

        # Run lvs command with JSON output including segment information
        result = run(f'lvs -a -o lv_all,seg_all {self.vg}/{self.name} --reportformat json')
        if result.failed or not result.stdout:
            logging.error(f'Failed to get LV report data for {self.vg}/{self.name}')
            return False

        try:
            report_data = json.loads(result.stdout)
            return self._update_from_report(report_data)
        except json.JSONDecodeError:
            logging.exception('Failed to parse LV report JSON')
            return False

    def _update_from_report(self, report_data: dict[str, Any]) -> bool:
        """Update LV information from report data.

        Args:
            report_data: Complete report data from lvs JSON

        Returns:
            bool: True if update was successful, False otherwise
        """
        if self.prevent_update:
            logging.debug('Update from report skipped due to prevent_update flag')
            return True

        if not isinstance(report_data, dict) or 'report' not in report_data:
            logging.error('Invalid LV report format')
            return False

        reports = report_data.get('report', [])
        if not isinstance(reports, list) or not reports:
            logging.error('No reports found in LV data')
            return False

        # Get the first report
        report = reports[0]
        if not isinstance(report, dict) or 'lv' not in report:
            logging.error('Invalid report structure')
            return False

        lvs = report.get('lv', [])
        if not isinstance(lvs, list) or not lvs:
            logging.warning(f'No LV data found for {self.vg}/{self.name}')
            return False

        # Get the first (and should be only) LV
        lv_data = lvs[0]
        if not isinstance(lv_data, dict):
            logging.error('Invalid LV data structure')
            return False

        # Update all fields from the data
        self.raw_data = lv_data.copy()

        # Map all known fields
        for field_name in self.__dataclass_fields__:
            if field_name not in ('raw_data', 'name', 'vg', 'prevent_update') and field_name in lv_data:
                setattr(self, field_name, lv_data[field_name])

        # Update our name and vg from the data if not set
        if not self.name and self.lv_name:
            self.name = self.lv_name
        if not self.vg and self.vg_name:
            self.vg = self.vg_name

        # Extract VG name from full name if available
        if self.lv_full_name and not self.vg_name and '/' in self.lv_full_name:
            self.vg_name = self.lv_full_name.split('/')[0]

        return True

    @classmethod
    def get_all(cls, vg: str | None = None) -> list[LVReport]:
        """Get reports for all logical volumes.

        Args:
            vg: Optional volume group to filter by

        Returns:
            List of LVReport instances
        """
        reports: list[LVReport] = []

        # Build command
        cmd = 'lvs -o lv_all,seg_all --reportformat json'
        if vg:
            cmd += f' {vg}'

        result = run(cmd)
        if result.failed or not result.stdout:
            return reports

        try:
            report_data = json.loads(result.stdout)

            if 'report' in report_data and isinstance(report_data['report'], list):
                for report in report_data['report']:
                    if not isinstance(report, dict) or 'lv' not in report:
                        continue

                    for lv_data in report.get('lv', []):
                        if not isinstance(lv_data, dict):
                            continue

                        # Create report with prevent_update=True to avoid double refresh
                        lv_report = cls(prevent_update=True)
                        lv_report.raw_data = lv_data.copy()

                        # Map all known fields
                        for field_name in cls.__dataclass_fields__:
                            if field_name not in ('raw_data', 'name', 'vg', 'prevent_update') and field_name in lv_data:
                                setattr(lv_report, field_name, lv_data[field_name])

                        # Set name and vg from the data
                        lv_report.name = lv_report.lv_name
                        lv_report.vg = lv_report.vg_name

                        # Extract VG name from full name if needed
                        if lv_report.lv_full_name and not lv_report.vg_name and '/' in lv_report.lv_full_name:
                            lv_report.vg_name = lv_report.lv_full_name.split('/')[0]
                            if not lv_report.vg:
                                lv_report.vg = lv_report.vg_name

                        reports.append(lv_report)
        except (json.JSONDecodeError, KeyError, ValueError) as e:
            logging.warning(f'Failed to parse LV reports: {e}')

        return reports

__post_init__()

Initialize the report.

Source code in sts_libs/src/sts/lvm.py 
293
294
295
296
297
def __post_init__(self) -> None:
    """Initialize the report."""
    # If name and vg are provided, fetch the report data
    if self.name and self.vg:
        self.refresh()

get_all(vg=None) classmethod

Get reports for all logical volumes.

Parameters:

Name Type Description Default
vg str | None

Optional volume group to filter by

 None

Returns:

Type Description
list[LVReport]

List of LVReport instances
Source code in sts_libs/src/sts/lvm.py 
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
@classmethod
def get_all(cls, vg: str | None = None) -> list[LVReport]:
    """Get reports for all logical volumes.

    Args:
        vg: Optional volume group to filter by

    Returns:
        List of LVReport instances
    """
    reports: list[LVReport] = []

    # Build command
    cmd = 'lvs -o lv_all,seg_all --reportformat json'
    if vg:
        cmd += f' {vg}'

    result = run(cmd)
    if result.failed or not result.stdout:
        return reports

    try:
        report_data = json.loads(result.stdout)

        if 'report' in report_data and isinstance(report_data['report'], list):
            for report in report_data['report']:
                if not isinstance(report, dict) or 'lv' not in report:
                    continue

                for lv_data in report.get('lv', []):
                    if not isinstance(lv_data, dict):
                        continue

                    # Create report with prevent_update=True to avoid double refresh
                    lv_report = cls(prevent_update=True)
                    lv_report.raw_data = lv_data.copy()

                    # Map all known fields
                    for field_name in cls.__dataclass_fields__:
                        if field_name not in ('raw_data', 'name', 'vg', 'prevent_update') and field_name in lv_data:
                            setattr(lv_report, field_name, lv_data[field_name])

                    # Set name and vg from the data
                    lv_report.name = lv_report.lv_name
                    lv_report.vg = lv_report.vg_name

                    # Extract VG name from full name if needed
                    if lv_report.lv_full_name and not lv_report.vg_name and '/' in lv_report.lv_full_name:
                        lv_report.vg_name = lv_report.lv_full_name.split('/')[0]
                        if not lv_report.vg:
                            lv_report.vg = lv_report.vg_name

                    reports.append(lv_report)
    except (json.JSONDecodeError, KeyError, ValueError) as e:
        logging.warning(f'Failed to parse LV reports: {e}')

    return reports

refresh()

Refresh LV report data from system.

Updates all fields with the latest information from lvs command.

Returns:

Name Type Description
bool bool

True if refresh was successful, False otherwise
Source code in sts_libs/src/sts/lvm.py 
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
def refresh(self) -> bool:
    """Refresh LV report data from system.

    Updates all fields with the latest information from lvs command.

    Returns:
        bool: True if refresh was successful, False otherwise
    """
    # If prevent_update is True, skip refresh
    if self.prevent_update:
        logging.debug('Refresh skipped due to prevent_update flag')
        return True

    if not self.name or not self.vg:
        logging.error('LV name and VG name required for refresh')
        return False

    # Run lvs command with JSON output including segment information
    result = run(f'lvs -a -o lv_all,seg_all {self.vg}/{self.name} --reportformat json')
    if result.failed or not result.stdout:
        logging.error(f'Failed to get LV report data for {self.vg}/{self.name}')
        return False

    try:
        report_data = json.loads(result.stdout)
        return self._update_from_report(report_data)
    except json.JSONDecodeError:
        logging.exception('Failed to parse LV report JSON')
        return False

LogicalVolume dataclass

Bases: LvmDevice

Logical Volume device.

A Logical Volume (LV) is a virtual partition created from VG space. LVs appear as block devices that can be formatted and mounted.

Key features: - Flexible sizing - Online resizing - Snapshots - Striping and mirroring - Thin provisioning

Parameters:

Name Type Description Default
name str | None

Device name (optional)

 None
path Path | str | None

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

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
model str | None

Device model (optional)

 None
yes bool

Automatically answer yes to prompts

 True
force bool

Force operations without confirmation

 False
vg str | None

Volume group name (optional, discovered from device)

 None
report LVReport | None

LV report instance (optional, created automatically)

 None
prevent_report_updates bool

Prevent automatic report updates (defaults to False)

 False

The LogicalVolume class now includes integrated report functionality: - Automatic report creation when name and vg are provided - Automatic report refresh after state-changing operations - Access to detailed LV information directly via report attributes - Prevention of updates via prevent_report_updates flag

Example 
# Basic usage with automatic report
lv = LogicalVolume(name='lv0', vg='vg0')
lv.create(size='100M')
print(lv.report.lv_size)

# Prevent automatic updates
lv = LogicalVolume(name='lv0', vg='vg0', prevent_report_updates=True)

# Manual report refresh
lv.refresh_report()
Source code in sts_libs/src/sts/lvm.py 
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
@dataclass
class LogicalVolume(LvmDevice):
    """Logical Volume device.

    A Logical Volume (LV) is a virtual partition created from VG space.
    LVs appear as block devices that can be formatted and mounted.

    Key features:
    - Flexible sizing
    - Online resizing
    - Snapshots
    - Striping and mirroring
    - Thin provisioning

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<vg>/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        vg: Volume group name (optional, discovered from device)
        report: LV report instance (optional, created automatically)
        prevent_report_updates: Prevent automatic report updates (defaults to False)

    The LogicalVolume class now includes integrated report functionality:
    - Automatic report creation when name and vg are provided
    - Automatic report refresh after state-changing operations
    - Access to detailed LV information directly via report attributes
    - Prevention of updates via prevent_report_updates flag

    Example:
        ```python
        # Basic usage with automatic report
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.create(size='100M')
        print(lv.report.lv_size)

        # Prevent automatic updates
        lv = LogicalVolume(name='lv0', vg='vg0', prevent_report_updates=True)

        # Manual report refresh
        lv.refresh_report()
        ```
    """

    # Optional parameters for this class
    vg: str | None = None  # Parent VG
    pool_name: str | None = None
    report: LVReport | None = field(default=None, repr=False)
    prevent_report_updates: bool = False

    # Available LV commands
    COMMANDS: ClassVar[list[str]] = [
        'lvchange',  # Change LV attributes
        'lvcreate',  # Create LV
        'lvconvert',  # Convert LV type
        'lvdisplay',  # Show LV details
        'lvextend',  # Increase LV size
        'lvreduce',  # Reduce LV size
        'lvremove',  # Remove LV
        'lvrename',  # Rename LV
        'lvresize',  # Change LV size
        'lvs',  # List LVs
        'lvscan',  # Scan for LVs
    ]

    def __post_init__(self) -> None:
        """Initialize Logical Volume.

        - Sets device path from name and VG
        - Discovers VG membership
        - Creates and updates from report
        """
        # Set path based on name and vg if not provided
        if not self.path and self.name and self.vg:
            self.path = f'/dev/{self.vg}/{self.name}'

        # Initialize parent class
        super().__post_init__()

    def refresh_report(self) -> bool:
        """Refresh LV report data.

        Creates or updates the LV report with the latest information.

        Returns:
            bool: True if refresh was successful
        """
        # Create new report if needed
        if not self.report:
            # Do not provide name and vg during init to prevent update
            self.report = LVReport()
            self.report.name = self.name
            self.report.vg = self.vg

        # Refresh the report data
        return self.report.refresh()

    def discover_vg(self) -> str | None:
        """Discover VG if name is available."""
        if self.name and not self.vg:
            result = run(f'lvs {self.name} -o vg_name --noheadings')
            if result.succeeded:
                self.vg = result.stdout.strip()
                return self.vg
        return None

    def create(self, *args: str, **options: str) -> bool:
        """Create Logical Volume.

        Creates a new LV in the specified VG:
        - Allocates space from VG
        - Creates device mapper device
        - Initializes LV metadata

        Args:
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.create(size='1G')
            True
            ```
        """
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.vg:
            logging.error('Volume group required')
            return False

        result = self._run('lvcreate', '-n', self.name, self.vg, *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def remove(self, *args: str, **options: str) -> bool:
        """Remove Logical Volume.

        Removes LV and its data:
        - Data is permanently lost
        - Space is returned to VG
        - Device mapper device is removed

        Args:
            *args: Additional volume paths to remove (for removing multiple volumes)
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Remove single volume
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.remove()
            True

            # Remove multiple volumes
            lv = LogicalVolume(name='lv1', vg='vg0')
            lv.remove('vg0/lv2', 'vg0/lv3')
            True
            ```
        """
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.vg:
            logging.error('Volume group required')
            return False

        # Start with this LV
        targets = [f'{self.vg}/{self.name}']

        # Add any additional volumes from args
        if args:
            targets.extend(args)

        result = self._run('lvremove', *targets, **options)
        return result.succeeded

    def change(self, *args: str, **options: str) -> bool:
        """Change Logical Volume attributes.

        Change a general LV attribute:

        Args:
            *args: LV options (see LVMOptions)
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.change('-an', 'vg0/lv0')
            True
            ```
        """
        result = self._run('lvchange', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def extend(self, **options: str) -> bool:
        """Extend Logical volume.

        - LV must be initialized (using lvcreate)
        - VG must have sufficient usable space

        Args:
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lvol0', vg='vg0')
            lv.extend(extents='100%vg')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvextend', f'{self.vg}/{self.name}', **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def lvs(self, *args: str, **options: str) -> CommandResult:
        """Get information about logical volumes.

        Executes the 'lvs' command with optional filtering to display
        information about logical volumes.

        Args:
            *args: Positional args passed through to `lvs` (e.g., LV selector, flags).
            **options: LV command options (see LvmOptions).

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume()
            result = lv.lvs()
            print(result.stdout)
            ```
        """
        return self._run('lvs', *args, **options)

    def convert(self, *args: str, **options: str) -> bool:
        """Convert Logical Volume type.

        Converts LV type (linear, striped, mirror, snapshot, etc):
        - Can change between different LV types
        - May require additional space or devices
        - Some conversions are irreversible

        Args:
            *args: LV conversion arguments
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.convert('--type', 'mirror')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_splitmirrors(self, count: int, new_name: str, **options: str) -> tuple[bool, LogicalVolume | None]:
        """Split images from a raid1 or mirror LV and create a new LV.

        Splits the specified number of images from a raid1 or mirror LV and uses them
        to create a new LV with the specified name.

        Args:
            count: Number of mirror images to split
            new_name: Name for the new LV created from split images
            **options: Additional LV options (see LvmOptions)

        Returns:
            Tuple of (success, new_lv) where:
            - success: True if successful, False otherwise
            - new_lv: LogicalVolume object for the new split LV, or None if failed

        Example:
            ```python
            lv = LogicalVolume(name='mirror_lv', vg='vg0')
            success, split_lv = lv.convert_splitmirrors(1, 'split_lv')
            if success:
                print(f'Created new LV: {split_lv.name}')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False, None
        if not self.name:
            logging.error('Logical volume name required')
            return False, None

        result = self._run(
            'lvconvert', '--splitmirrors', str(count), '--name', new_name, f'{self.vg}/{self.name}', **options
        )
        success = result.succeeded

        if success:
            self.refresh_report()

        # Create LogicalVolume object for the new split LV
        new_lv = None
        if success:
            try:
                new_lv = LogicalVolume(name=new_name, vg=self.vg)
                if not new_lv.refresh_report():
                    logging.warning(f'Failed to refresh report for new LV {new_name}')
            except (ValueError, OSError) as e:
                logging.warning(f'Failed to create LogicalVolume object for {new_name}: {e}')
                new_lv = None

        return success, new_lv

    def convert_to_thinpool(self, **options: str) -> bool:
        """Convert logical volume to thin pool.

        Converts an existing LV to a thin pool using lvconvert --thinpool.
        The LV must already exist and have sufficient space.

        Args:
            **options: Conversion options including:
                - chunksize: Chunk size for the thin pool (e.g., '256k')
                - zero: Whether to zero the first 4KiB ('y' or 'n')
                - discards: Discard policy ('passdown', 'nopassdown', 'ignore')
                - poolmetadatasize: Size of pool metadata (e.g., '4M')
                - poolmetadata: Name of existing LV to use as metadata
                - readahead: Read-ahead value
                - Other lvconvert options

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Basic conversion
            lv = LogicalVolume(name='pool', vg='vg0')
            lv.create(size='100M')
            lv.convert_to_thinpool()

            # Conversion with parameters
            lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

            # Conversion with separate metadata LV
            lv.convert_to_thinpool(poolmetadata='metadata_lv')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        # Build the lvconvert command
        args = ['--thinpool', f'{self.vg}/{self.name}']

        # Handle special parameter mappings
        if 'chunksize' in options:
            args.extend(['-c', options.pop('chunksize')])
        if 'zero' in options:
            zero_val = options.pop('zero')
            # Convert boolean to string if needed
            if isinstance(zero_val, bool):
                zero_val = 'y' if zero_val else 'n'
            args.extend(['-Z', zero_val])
        if 'readahead' in options:
            readahead_val = options.pop('readahead')
            # Convert numeric to string if needed
            args.extend(['-r', str(readahead_val)])

        result = self._run('lvconvert', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_pool_data(self, **options: str) -> bool:
        """Convert thin pool data component to specified type.

        Converts the thin pool's data component (e.g., from linear to RAID1).
        This is typically used to add mirroring or change the RAID level of the pool's data.

        Args:
            **options: Conversion options including:
                - type: Target type (e.g., 'raid1')
                - mirrors: Number of mirrors for RAID1
                - Other lvconvert options

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Convert pool data to RAID1 with 3 mirrors
            pool.convert_pool_data(type='raid1', mirrors='3')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.report or not self.report.data_lv:
            logging.error('Pool data LV not found - is this a thin pool?')
            return False

        # Convert the pool's data component
        result = self._run('lvconvert', f'{self.vg}/{self.name}_tdata', **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_pool_metadata(self, **options: str) -> bool:
        """Convert thin pool metadata component to specified type.

        Converts the thin pool's metadata component (e.g., from linear to RAID1).
        This is typically used to add mirroring or change the RAID level of the pool's metadata.

        Args:
            **options: Conversion options including:
                - type: Target type (e.g., 'raid1')
                - mirrors: Number of mirrors for RAID1
                - Other lvconvert options

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            # Convert pool metadata to RAID1 with 1 mirror
            pool.convert_pool_metadata(type='raid1', mirrors='1')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not self.report or not self.report.metadata_lv:
            logging.error('Pool metadata LV not found - is this a thin pool?')
            return False

        # Convert the pool's metadata component
        result = self._run('lvconvert', f'{self.vg}/{self.name}_tmeta', **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def convert_originname(self, thinpool: str, origin_name: str, **options: str) -> tuple[bool, LogicalVolume | None]:
        """Convert LV to thin LV with named external origin.

        Converts the LV to a thin LV in the specified thin pool, using the original LV
        as an external read-only origin with the specified name.

        Args:
            thinpool: Name of the thin pool (format: vg/pool or just pool if same VG)
            origin_name: Name for the external origin LV
            **options: Additional LV options (see LvmOptions)

        Returns:
            Tuple of (success, origin_lv) where:
            - success: True if successful, False otherwise
            - origin_lv: LogicalVolume object for the read-only origin LV, or None if failed

        Example:
            ```python
            lv = LogicalVolume(name='data_lv', vg='vg0')
            success, origin_lv = lv.convert_originname('vg0/thin_pool', 'data_origin')
            if success:
                print(f'Created origin LV: {origin_lv.name}')
                print(f'Original LV converted to thin LV')
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False, None
        if not self.name:
            logging.error('Logical volume name required')
            return False, None

        # Ensure thinpool has VG prefix if not provided
        if '/' not in thinpool:
            thinpool = f'{self.vg}/{thinpool}'

        result = self._run(
            'lvconvert',
            '--type',
            'thin',
            '--thinpool',
            thinpool,
            '--originname',
            origin_name,
            f'{self.vg}/{self.name}',
            **options,
        )
        success = result.succeeded

        if success:
            self.refresh_report()

        # Create LogicalVolume object for the origin LV
        origin_lv = None
        if success:
            try:
                origin_lv = LogicalVolume(name=origin_name, vg=self.vg)
                if not origin_lv.refresh_report():
                    logging.warning(f'Failed to refresh report for origin LV {origin_name}')
            except (ValueError, OSError) as e:
                logging.warning(f'Failed to create LogicalVolume object for {origin_name}: {e}')
                origin_lv = None

        return success, origin_lv

    def display(self, **options: str) -> CommandResult:
        """Display Logical Volume details.

        Shows detailed information about the LV:
        - Size and allocation
        - Attributes and permissions
        - Segment information
        - Device mapper details

        Args:
            **options: LV options (see LvmOptions)

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            result = lv.display()
            print(result.stdout)
            ```
        """
        if not self.vg or not self.name:
            return self._run('lvdisplay', **options)
        return self._run('lvdisplay', f'{self.vg}/{self.name}', **options)

    def reduce(self, *args: str, **options: str) -> bool:
        """Reduce Logical Volume size.

        Reduces LV size (shrinks the volume):
        - Filesystem must be shrunk first
        - Data loss risk if not done carefully
        - Cannot reduce below used space

        Args:
            *args: Additional lvreduce arguments
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.reduce(size='500M')
            True

            # With additional arguments
            lv.reduce('--test', size='500M')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        result = self._run('lvreduce', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def rename(self, new_name: str, **options: str) -> bool:
        """Rename Logical Volume.

        Changes the LV name:
        - Must not conflict with existing LV names
        - Updates device mapper devices
        - May require remounting if mounted

        Args:
            new_name: New name for the LV
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.rename('new_lv')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False
        if not new_name:
            logging.error('New name required')
            return False

        result = self._run('lvrename', f'{self.vg}/{self.name}', new_name, **options)
        if result.succeeded:
            self.name = new_name
            self.path = f'/dev/{self.vg}/{self.name}'
            self.refresh_report()
        return result.succeeded

    def resize(self, *args: str, **options: str) -> bool:
        """Resize Logical Volume.

        Changes LV size (can grow or shrink):
        - Combines extend and reduce functionality
        - Safer than lvreduce for shrinking
        - Can resize filesystem simultaneously

        Args:
            *args: Additional lvresize arguments (e.g., '-l+2', '-t', '--test')
            **options: LV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            lv = LogicalVolume(name='lv0', vg='vg0')
            lv.resize(size='2G')
            True

            # With additional arguments
            lv.resize('-l+2', size='2G')
            True

            # With test flag
            lv.resize('--test', size='2G')
            True
            ```
        """
        if not self.vg:
            logging.error('Volume group name required')
            return False
        if not self.name:
            logging.error('Logical volume name required')
            return False

        result = self._run('lvresize', f'{self.vg}/{self.name}', *args, **options)
        if result.succeeded:
            self.refresh_report()
        return result.succeeded

    def scan(self, *args: str, **options: str) -> CommandResult:
        """Scan for Logical Volumes.

        Scans all devices for LV information:
        - Discovers new LVs
        - Updates device mapper
        - Useful after system changes

        Args:
            **options: LV options (see LvmOptions)

        Returns:
            CommandResult object containing command output and status

        Example:
            ```python
            lv = LogicalVolume()
            result = lv.scan()
            print(result.stdout)
            ```
        """
        return self._run('lvscan', *args, **options)

    def deactivate(self) -> bool:
        """Deactivate Logical Volume."""
        udevadm_settle()
        result = self.change('-an', f'{self.vg}/{self.name}')
        udevadm_settle()
        if result:
            return self.wait_for_lv_deactivation()
        return result

    def activate(self) -> bool:
        """Activate Logical Volume."""
        return self.change('-ay', f'{self.vg}/{self.name}')

    def wait_for_lv_deactivation(self, timeout: int = 30) -> bool:
        """Wait for logical volume to be fully deactivated.

        Args:
            timeout: Maximum wait time in seconds

        Returns:
            True if deactivated successfully, False if timeout
        """
        start_time = time.time()

        while time.time() - start_time < timeout:
            # Check LV status using lvs command
            result = self.lvs(f'{self.vg}/{self.name}', '--noheadings', '-o lv_active')
            if result.succeeded and 'active' not in result.stdout.lower():
                # LV is inactive - also verify device node is gone
                if self.path is not None:
                    device_path = Path(self.path)
                    if not device_path.exists():
                        return True
                else:
                    return True  # If no path, consider it deactivated
            time.sleep(2)  # Poll every 2 seconds

        logging.warning(f'LV {self.vg}/{self.name} deactivation timed out after {timeout}s')
        return False

    def change_discards(self, discards_value: str) -> bool:
        """Change discards setting for logical volume.

        Args:
            discards_value: New discards value ('passdown', 'nopassdown', 'ignore')

        Returns:
            True if change succeeded
        """
        if not self.vg or not self.name:
            logging.error('Volume group and logical volume name required')
            return False

        return self.change('--discards', discards_value, f'{self.vg}/{self.name}')

    def create_snapshot(self, snapshot_name: str, *args: str, **options: str) -> LogicalVolume | None:
        """Create snapshot of this LV.

        Creates a snapshot of the current LV (self) with the given snapshot name.
        For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

        Args:
            snapshot_name: Name for the new snapshot LV
            *args: Additional command-line arguments (e.g., '-K', '-k', '--test')
            **options: Snapshot options (see LvmOptions)

        Returns:
            LogicalVolume instance representing the created snapshot, or None if failed

        Example:
            ```python
            # Create origin LV and then snapshot it
            origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
            # ... create the origin LV ...
            # Create thin snapshot (no size needed)
            snap1 = origin_lv.create_snapshot('snap1')

            # Create COW snapshot with ignore activation skip
            snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')
            ```
        """
        if not self.name:
            logging.error('Origin LV name required')
            return None
        if not self.vg:
            logging.error('Volume group required')
            return None
        if not snapshot_name:
            logging.error('Snapshot name required')
            return None

        # Build snapshot command
        cmd_args = ['-s']

        # Add any additional arguments
        if args:
            cmd_args.extend(args)

        # Add origin (this LV)
        cmd_args.append(f'{self.vg}/{self.name}')

        # Add snapshot name
        cmd_args.extend(['-n', snapshot_name])

        result = self._run('lvcreate', *cmd_args, **options)
        if result.succeeded:
            self.refresh_report()
            # Return new LogicalVolume instance representing the snapshot
            snap = LogicalVolume(name=snapshot_name, vg=self.vg)
            snap.refresh_report()
            return snap
        return None

    def get_pool_usage(self) -> tuple[str, str]:
        """Get thin pool data and metadata usage percentages.

        Returns:
            Tuple of (data_percent, metadata_percent) as strings
        """
        if not self.vg or not self.name:
            logging.error('Volume group and logical volume name required')
            return 'unknown', 'unknown'

        # Get data usage
        data_result = self.lvs(f'{self.vg}/{self.name}', o='data_percent', noheadings='')
        data_percent = data_result.stdout.strip() if data_result.succeeded else 'unknown'

        # Get metadata usage
        meta_result = self.lvs(f'{self.vg}/{self.name}', o='metadata_percent', noheadings='')
        meta_percent = meta_result.stdout.strip() if meta_result.succeeded else 'unknown'

        return data_percent, meta_percent

    @classmethod
    def create_thin_pool(cls, pool_name: str, vg_name: str, **options: str) -> LogicalVolume:
        """Create thin pool with specified options.

        Args:
            pool_name: Pool name
            vg_name: Volume group name
            **options: Pool creation options

        Returns:
            LogicalVolume object for the created pool

        Raises:
            AssertionError: If pool creation fails
        """
        pool = cls(name=pool_name, vg=vg_name)
        options['type'] = 'thin-pool'
        assert pool.create(**options), f'Failed to create thin pool {pool_name}'
        return pool

    @classmethod
    def create_thin_volume(
        cls, lv_name: str, vg_name: str, pool_name: str, virtualsize: str, **options: str
    ) -> LogicalVolume:
        """Create thin volume in specified pool.

        Args:
            lv_name: Thin volume name
            vg_name: Volume group name
            pool_name: Parent pool name
            virtualsize: Virtual size for thin volume
            **options: Thin volume creation options

        Returns:
            LogicalVolume object for the created thin volume

        Raises:
            AssertionError: If volume creation fails
        """
        thin_lv = cls(name=lv_name, pool_name=pool_name, vg=vg_name)
        assert thin_lv.create(virtualsize=virtualsize, type='thin', thinpool=pool_name, **options), (
            f'Failed to create thin volume {lv_name}'
        )
        return thin_lv

    @classmethod
    def from_report(cls, report: LVReport) -> LogicalVolume | None:
        """Create LogicalVolume from LVReport.

        Args:
            report: LV report data

        Returns:
            LogicalVolume instance or None if invalid

        Example:
            ```python
            lv = LogicalVolume.from_report(report)
            ```
        """
        if not report.name or not report.vg:
            return None

        # Create LogicalVolume with report already attached
        return cls(
            name=report.name,
            vg=report.vg,
            path=report.lv_path,
            report=report,  # Attach the report directly
            prevent_report_updates=True,  # Avoid double refresh since report is already fresh
        )

    @classmethod
    def get_all(cls, vg: str | None = None) -> list[LogicalVolume]:
        """Get all Logical Volumes.

        Args:
            vg: Optional volume group to filter by

        Returns:
            List of LogicalVolume instances

        Example:
            ```python
            LogicalVolume.get_all()
            [LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]
            ```
        """
        logical_volumes: list[LogicalVolume] = []

        # Get all reports
        reports = LVReport.get_all(vg)

        # Create LogicalVolumes from reports
        logical_volumes.extend(lv for report in reports if (lv := cls.from_report(report)))

        return logical_volumes

    def get_data_stripes(self) -> str | None:
        """Get stripe count for thin pool data component.

        For thin pools, the actual stripe information is stored in the data component (_tdata),
        not in the main pool LV. This method returns the stripe count from the data component.

        Returns:
            String representing stripe count, or None if not a thin pool or stripes not found

        Example:
            ```python
            pool = LogicalVolume(name='pool', vg='vg0')
            pool.create(stripes='2', size='100M', type='thin-pool')
            stripe_count = pool.get_data_stripes()  # Returns '2'
            ```
        """
        if not self.report or not self.report.data_lv:
            # Not a thin pool or no data component
            return None

        # Get the data component name (usually pool_name + '_tdata')
        data_lv_name = self.report.data_lv
        if not data_lv_name:
            return None

        # Remove brackets if present (e.g., [pool1_tdata] -> pool1_tdata)
        data_lv_name = data_lv_name.strip('[]')

        # Create LogicalVolume for the data component and get its stripes
        try:
            data_lv = LogicalVolume(name=data_lv_name, vg=self.vg)
            if data_lv.refresh_report() and data_lv.report:
                return data_lv.report.stripes
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to get stripe info from data component {data_lv_name}: {e}')

        return None

    def get_data_stripe_size(self) -> str | None:
        """Get stripe size for thin pool data component.

        For thin pools, the actual stripe size information is stored in the data component (_tdata),
        not in the main pool LV. This method returns the stripe size from the data component.

        Returns:
            String representing stripe size, or None if not a thin pool or stripe size not found

        Example:
            ```python
            pool = LogicalVolume(name='pool', vg='vg0')
            pool.create(stripes='2', stripesize='64k', size='100M', type='thin-pool')
            stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'
            ```
        """
        if not self.report or not self.report.data_lv:
            # Not a thin pool or no data component
            return None

        # Get the data component name (usually pool_name + '_tdata')
        data_lv_name = self.report.data_lv
        if not data_lv_name:
            return None

        # Remove brackets if present (e.g., [pool1_tdata] -> pool1_tdata)
        data_lv_name = data_lv_name.strip('[]')

        # Create LogicalVolume for the data component and get its stripe size
        try:
            data_lv = LogicalVolume(name=data_lv_name, vg=self.vg)
            if data_lv.refresh_report() and data_lv.report:
                return data_lv.report.stripe_size
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to get stripe size info from data component {data_lv_name}: {e}')

        return None

    def __eq__(self, other: object) -> bool:
        """Compare two LogicalVolume instances for equality.

        Two LogicalVolume instances are considered equal if they have the same:
        - name
        - volume group (vg)
        - pool_name (if both have a pool_name)

        For thin LVs that belong to a pool, the pool_name must also match.
        For regular LVs or when either LV has no pool_name, only name and vg are compared.

        Args:
            other: Object to compare with

        Returns:
            True if the LogicalVolume instances are equal, False otherwise

        Example:
            ```python
            lv1 = LogicalVolume(name='lv0', vg='vg0')
            lv2 = LogicalVolume(name='lv0', vg='vg0')
            lv3 = LogicalVolume(name='lv1', vg='vg0')

            assert lv1 == lv2  # Same name and vg
            assert lv1 != lv3  # Different name

            # For thin LVs with pools
            thin1 = LogicalVolume(name='thin1', vg='vg0')
            thin1.pool_name = 'pool'
            thin2 = LogicalVolume(name='thin1', vg='vg0')
            thin2.pool_name = 'pool'
            thin3 = LogicalVolume(name='thin1', vg='vg0')
            thin3.pool_name = 'other_pool'

            assert thin1 == thin2  # Same name, vg, and pool
            assert thin1 != thin3  # Different pool
            ```
        """
        if not isinstance(other, LogicalVolume):
            return False
        if self.pool_name is None or other.pool_name is None:
            return self.name == other.name and self.vg == other.vg
        return self.name == other.name and self.vg == other.vg and self.pool_name == other.pool_name

    def __hash__(self) -> int:
        """Generate hash for LogicalVolume instance.

        The hash is based on the combination of name, volume group (vg), and pool_name.
        This allows LogicalVolume instances to be used in sets and as dictionary keys.
        The hash is consistent with the equality comparison in __eq__.

        Returns:
            int: Hash value based on (name, vg, pool_name) tuple

        Example:
            ```python
            lv1 = LogicalVolume(name='lv0', vg='vg0')
            lv2 = LogicalVolume(name='lv0', vg='vg0')

            # Can be used in sets
            lv_set = {lv1, lv2}  # Only one instance since they're equal

            # Can be used as dictionary keys
            lv_dict = {lv1: 'some_value'}
            ```
        """
        return hash((self.name, self.vg, self.pool_name))

__eq__(other)

Compare two LogicalVolume instances for equality.

Two LogicalVolume instances are considered equal if they have the same: - name - volume group (vg) - pool_name (if both have a pool_name)

For thin LVs that belong to a pool, the pool_name must also match. For regular LVs or when either LV has no pool_name, only name and vg are compared.

Parameters:

Name Type Description Default
other object

Object to compare with

 required

Returns:

Type Description
bool

True if the LogicalVolume instances are equal, False otherwise
Example 
lv1 = LogicalVolume(name='lv0', vg='vg0')
lv2 = LogicalVolume(name='lv0', vg='vg0')
lv3 = LogicalVolume(name='lv1', vg='vg0')

assert lv1 == lv2  # Same name and vg
assert lv1 != lv3  # Different name

# For thin LVs with pools
thin1 = LogicalVolume(name='thin1', vg='vg0')
thin1.pool_name = 'pool'
thin2 = LogicalVolume(name='thin1', vg='vg0')
thin2.pool_name = 'pool'
thin3 = LogicalVolume(name='thin1', vg='vg0')
thin3.pool_name = 'other_pool'

assert thin1 == thin2  # Same name, vg, and pool
assert thin1 != thin3  # Different pool
Source code in sts_libs/src/sts/lvm.py 
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
def __eq__(self, other: object) -> bool:
    """Compare two LogicalVolume instances for equality.

    Two LogicalVolume instances are considered equal if they have the same:
    - name
    - volume group (vg)
    - pool_name (if both have a pool_name)

    For thin LVs that belong to a pool, the pool_name must also match.
    For regular LVs or when either LV has no pool_name, only name and vg are compared.

    Args:
        other: Object to compare with

    Returns:
        True if the LogicalVolume instances are equal, False otherwise

    Example:
        ```python
        lv1 = LogicalVolume(name='lv0', vg='vg0')
        lv2 = LogicalVolume(name='lv0', vg='vg0')
        lv3 = LogicalVolume(name='lv1', vg='vg0')

        assert lv1 == lv2  # Same name and vg
        assert lv1 != lv3  # Different name

        # For thin LVs with pools
        thin1 = LogicalVolume(name='thin1', vg='vg0')
        thin1.pool_name = 'pool'
        thin2 = LogicalVolume(name='thin1', vg='vg0')
        thin2.pool_name = 'pool'
        thin3 = LogicalVolume(name='thin1', vg='vg0')
        thin3.pool_name = 'other_pool'

        assert thin1 == thin2  # Same name, vg, and pool
        assert thin1 != thin3  # Different pool
        ```
    """
    if not isinstance(other, LogicalVolume):
        return False
    if self.pool_name is None or other.pool_name is None:
        return self.name == other.name and self.vg == other.vg
    return self.name == other.name and self.vg == other.vg and self.pool_name == other.pool_name

__hash__()

Generate hash for LogicalVolume instance.

The hash is based on the combination of name, volume group (vg), and pool_name. This allows LogicalVolume instances to be used in sets and as dictionary keys. The hash is consistent with the equality comparison in eq.

Returns:

Name Type Description
int int

Hash value based on (name, vg, pool_name) tuple
Example 
lv1 = LogicalVolume(name='lv0', vg='vg0')
lv2 = LogicalVolume(name='lv0', vg='vg0')

# Can be used in sets
lv_set = {lv1, lv2}  # Only one instance since they're equal

# Can be used as dictionary keys
lv_dict = {lv1: 'some_value'}
Source code in sts_libs/src/sts/lvm.py 
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
def __hash__(self) -> int:
    """Generate hash for LogicalVolume instance.

    The hash is based on the combination of name, volume group (vg), and pool_name.
    This allows LogicalVolume instances to be used in sets and as dictionary keys.
    The hash is consistent with the equality comparison in __eq__.

    Returns:
        int: Hash value based on (name, vg, pool_name) tuple

    Example:
        ```python
        lv1 = LogicalVolume(name='lv0', vg='vg0')
        lv2 = LogicalVolume(name='lv0', vg='vg0')

        # Can be used in sets
        lv_set = {lv1, lv2}  # Only one instance since they're equal

        # Can be used as dictionary keys
        lv_dict = {lv1: 'some_value'}
        ```
    """
    return hash((self.name, self.vg, self.pool_name))

__post_init__()

Initialize Logical Volume.

  • Sets device path from name and VG
  • Discovers VG membership
  • Creates and updates from report
Source code in sts_libs/src/sts/lvm.py 
973
974
975
976
977
978
979
980
981
982
983
984
985
def __post_init__(self) -> None:
    """Initialize Logical Volume.

    - Sets device path from name and VG
    - Discovers VG membership
    - Creates and updates from report
    """
    # Set path based on name and vg if not provided
    if not self.path and self.name and self.vg:
        self.path = f'/dev/{self.vg}/{self.name}'

    # Initialize parent class
    super().__post_init__()

activate()

Activate Logical Volume.

Source code in sts_libs/src/sts/lvm.py 
1632
1633
1634
def activate(self) -> bool:
    """Activate Logical Volume."""
    return self.change('-ay', f'{self.vg}/{self.name}')

change(*args, **options)

Change Logical Volume attributes.

Change a general LV attribute:

Parameters:

Name Type Description Default
*args str

LV options (see LVMOptions)

 ()
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.change('-an', 'vg0/lv0')
True
Source code in sts_libs/src/sts/lvm.py 
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
def change(self, *args: str, **options: str) -> bool:
    """Change Logical Volume attributes.

    Change a general LV attribute:

    Args:
        *args: LV options (see LVMOptions)
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.change('-an', 'vg0/lv0')
        True
        ```
    """
    result = self._run('lvchange', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

change_discards(discards_value)

Change discards setting for logical volume.

Parameters:

Name Type Description Default
discards_value str

New discards value ('passdown', 'nopassdown', 'ignore')

 required

Returns:

Type Description
bool

True if change succeeded
Source code in sts_libs/src/sts/lvm.py 
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
def change_discards(self, discards_value: str) -> bool:
    """Change discards setting for logical volume.

    Args:
        discards_value: New discards value ('passdown', 'nopassdown', 'ignore')

    Returns:
        True if change succeeded
    """
    if not self.vg or not self.name:
        logging.error('Volume group and logical volume name required')
        return False

    return self.change('--discards', discards_value, f'{self.vg}/{self.name}')

convert(*args, **options)

Convert Logical Volume type.

Converts LV type (linear, striped, mirror, snapshot, etc): - Can change between different LV types - May require additional space or devices - Some conversions are irreversible

Parameters:

Name Type Description Default
*args str

LV conversion arguments

 ()
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.convert('--type', 'mirror')
True
Source code in sts_libs/src/sts/lvm.py 
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
def convert(self, *args: str, **options: str) -> bool:
    """Convert Logical Volume type.

    Converts LV type (linear, striped, mirror, snapshot, etc):
    - Can change between different LV types
    - May require additional space or devices
    - Some conversions are irreversible

    Args:
        *args: LV conversion arguments
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.convert('--type', 'mirror')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvconvert', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

convert_originname(thinpool, origin_name, **options)

Convert LV to thin LV with named external origin.

Converts the LV to a thin LV in the specified thin pool, using the original LV as an external read-only origin with the specified name.

Parameters:

Name Type Description Default
thinpool str

Name of the thin pool (format: vg/pool or just pool if same VG)

 required
origin_name str

Name for the external origin LV

 required
**options str

Additional LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

Tuple of (success, origin_lv) where:
LogicalVolume | None
  • success: True if successful, False otherwise
tuple[bool, LogicalVolume | None]
  • origin_lv: LogicalVolume object for the read-only origin LV, or None if failed
Example 
lv = LogicalVolume(name='data_lv', vg='vg0')
success, origin_lv = lv.convert_originname('vg0/thin_pool', 'data_origin')
if success:
    print(f'Created origin LV: {origin_lv.name}')
    print(f'Original LV converted to thin LV')
Source code in sts_libs/src/sts/lvm.py 
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
def convert_originname(self, thinpool: str, origin_name: str, **options: str) -> tuple[bool, LogicalVolume | None]:
    """Convert LV to thin LV with named external origin.

    Converts the LV to a thin LV in the specified thin pool, using the original LV
    as an external read-only origin with the specified name.

    Args:
        thinpool: Name of the thin pool (format: vg/pool or just pool if same VG)
        origin_name: Name for the external origin LV
        **options: Additional LV options (see LvmOptions)

    Returns:
        Tuple of (success, origin_lv) where:
        - success: True if successful, False otherwise
        - origin_lv: LogicalVolume object for the read-only origin LV, or None if failed

    Example:
        ```python
        lv = LogicalVolume(name='data_lv', vg='vg0')
        success, origin_lv = lv.convert_originname('vg0/thin_pool', 'data_origin')
        if success:
            print(f'Created origin LV: {origin_lv.name}')
            print(f'Original LV converted to thin LV')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False, None
    if not self.name:
        logging.error('Logical volume name required')
        return False, None

    # Ensure thinpool has VG prefix if not provided
    if '/' not in thinpool:
        thinpool = f'{self.vg}/{thinpool}'

    result = self._run(
        'lvconvert',
        '--type',
        'thin',
        '--thinpool',
        thinpool,
        '--originname',
        origin_name,
        f'{self.vg}/{self.name}',
        **options,
    )
    success = result.succeeded

    if success:
        self.refresh_report()

    # Create LogicalVolume object for the origin LV
    origin_lv = None
    if success:
        try:
            origin_lv = LogicalVolume(name=origin_name, vg=self.vg)
            if not origin_lv.refresh_report():
                logging.warning(f'Failed to refresh report for origin LV {origin_name}')
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to create LogicalVolume object for {origin_name}: {e}')
            origin_lv = None

    return success, origin_lv

convert_pool_data(**options)

Convert thin pool data component to specified type.

Converts the thin pool's data component (e.g., from linear to RAID1). This is typically used to add mirroring or change the RAID level of the pool's data.

Parameters:

Name Type Description Default
**options str

Conversion options including: - type: Target type (e.g., 'raid1') - mirrors: Number of mirrors for RAID1 - Other lvconvert options

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
# Convert pool data to RAID1 with 3 mirrors
pool.convert_pool_data(type='raid1', mirrors='3')
Source code in sts_libs/src/sts/lvm.py 
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
def convert_pool_data(self, **options: str) -> bool:
    """Convert thin pool data component to specified type.

    Converts the thin pool's data component (e.g., from linear to RAID1).
    This is typically used to add mirroring or change the RAID level of the pool's data.

    Args:
        **options: Conversion options including:
            - type: Target type (e.g., 'raid1')
            - mirrors: Number of mirrors for RAID1
            - Other lvconvert options

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Convert pool data to RAID1 with 3 mirrors
        pool.convert_pool_data(type='raid1', mirrors='3')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.report or not self.report.data_lv:
        logging.error('Pool data LV not found - is this a thin pool?')
        return False

    # Convert the pool's data component
    result = self._run('lvconvert', f'{self.vg}/{self.name}_tdata', **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

convert_pool_metadata(**options)

Convert thin pool metadata component to specified type.

Converts the thin pool's metadata component (e.g., from linear to RAID1). This is typically used to add mirroring or change the RAID level of the pool's metadata.

Parameters:

Name Type Description Default
**options str

Conversion options including: - type: Target type (e.g., 'raid1') - mirrors: Number of mirrors for RAID1 - Other lvconvert options

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
# Convert pool metadata to RAID1 with 1 mirror
pool.convert_pool_metadata(type='raid1', mirrors='1')
Source code in sts_libs/src/sts/lvm.py 
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
def convert_pool_metadata(self, **options: str) -> bool:
    """Convert thin pool metadata component to specified type.

    Converts the thin pool's metadata component (e.g., from linear to RAID1).
    This is typically used to add mirroring or change the RAID level of the pool's metadata.

    Args:
        **options: Conversion options including:
            - type: Target type (e.g., 'raid1')
            - mirrors: Number of mirrors for RAID1
            - Other lvconvert options

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Convert pool metadata to RAID1 with 1 mirror
        pool.convert_pool_metadata(type='raid1', mirrors='1')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.report or not self.report.metadata_lv:
        logging.error('Pool metadata LV not found - is this a thin pool?')
        return False

    # Convert the pool's metadata component
    result = self._run('lvconvert', f'{self.vg}/{self.name}_tmeta', **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

convert_splitmirrors(count, new_name, **options)

Split images from a raid1 or mirror LV and create a new LV.

Splits the specified number of images from a raid1 or mirror LV and uses them to create a new LV with the specified name.

Parameters:

Name Type Description Default
count int

Number of mirror images to split

 required
new_name str

Name for the new LV created from split images

 required
**options str

Additional LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

Tuple of (success, new_lv) where:
LogicalVolume | None
  • success: True if successful, False otherwise
tuple[bool, LogicalVolume | None]
  • new_lv: LogicalVolume object for the new split LV, or None if failed
Example 
lv = LogicalVolume(name='mirror_lv', vg='vg0')
success, split_lv = lv.convert_splitmirrors(1, 'split_lv')
if success:
    print(f'Created new LV: {split_lv.name}')
Source code in sts_libs/src/sts/lvm.py 
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
def convert_splitmirrors(self, count: int, new_name: str, **options: str) -> tuple[bool, LogicalVolume | None]:
    """Split images from a raid1 or mirror LV and create a new LV.

    Splits the specified number of images from a raid1 or mirror LV and uses them
    to create a new LV with the specified name.

    Args:
        count: Number of mirror images to split
        new_name: Name for the new LV created from split images
        **options: Additional LV options (see LvmOptions)

    Returns:
        Tuple of (success, new_lv) where:
        - success: True if successful, False otherwise
        - new_lv: LogicalVolume object for the new split LV, or None if failed

    Example:
        ```python
        lv = LogicalVolume(name='mirror_lv', vg='vg0')
        success, split_lv = lv.convert_splitmirrors(1, 'split_lv')
        if success:
            print(f'Created new LV: {split_lv.name}')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False, None
    if not self.name:
        logging.error('Logical volume name required')
        return False, None

    result = self._run(
        'lvconvert', '--splitmirrors', str(count), '--name', new_name, f'{self.vg}/{self.name}', **options
    )
    success = result.succeeded

    if success:
        self.refresh_report()

    # Create LogicalVolume object for the new split LV
    new_lv = None
    if success:
        try:
            new_lv = LogicalVolume(name=new_name, vg=self.vg)
            if not new_lv.refresh_report():
                logging.warning(f'Failed to refresh report for new LV {new_name}')
        except (ValueError, OSError) as e:
            logging.warning(f'Failed to create LogicalVolume object for {new_name}: {e}')
            new_lv = None

    return success, new_lv

convert_to_thinpool(**options)

Convert logical volume to thin pool.

Converts an existing LV to a thin pool using lvconvert --thinpool. The LV must already exist and have sufficient space.

Parameters:

Name Type Description Default
**options str

Conversion options including: - chunksize: Chunk size for the thin pool (e.g., '256k') - zero: Whether to zero the first 4KiB ('y' or 'n') - discards: Discard policy ('passdown', 'nopassdown', 'ignore') - poolmetadatasize: Size of pool metadata (e.g., '4M') - poolmetadata: Name of existing LV to use as metadata - readahead: Read-ahead value - Other lvconvert options

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
# Basic conversion
lv = LogicalVolume(name='pool', vg='vg0')
lv.create(size='100M')
lv.convert_to_thinpool()

# Conversion with parameters
lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

# Conversion with separate metadata LV
lv.convert_to_thinpool(poolmetadata='metadata_lv')
Source code in sts_libs/src/sts/lvm.py 
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
def convert_to_thinpool(self, **options: str) -> bool:
    """Convert logical volume to thin pool.

    Converts an existing LV to a thin pool using lvconvert --thinpool.
    The LV must already exist and have sufficient space.

    Args:
        **options: Conversion options including:
            - chunksize: Chunk size for the thin pool (e.g., '256k')
            - zero: Whether to zero the first 4KiB ('y' or 'n')
            - discards: Discard policy ('passdown', 'nopassdown', 'ignore')
            - poolmetadatasize: Size of pool metadata (e.g., '4M')
            - poolmetadata: Name of existing LV to use as metadata
            - readahead: Read-ahead value
            - Other lvconvert options

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Basic conversion
        lv = LogicalVolume(name='pool', vg='vg0')
        lv.create(size='100M')
        lv.convert_to_thinpool()

        # Conversion with parameters
        lv.convert_to_thinpool(chunksize='256k', zero='y', discards='nopassdown', poolmetadatasize='4M')

        # Conversion with separate metadata LV
        lv.convert_to_thinpool(poolmetadata='metadata_lv')
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    # Build the lvconvert command
    args = ['--thinpool', f'{self.vg}/{self.name}']

    # Handle special parameter mappings
    if 'chunksize' in options:
        args.extend(['-c', options.pop('chunksize')])
    if 'zero' in options:
        zero_val = options.pop('zero')
        # Convert boolean to string if needed
        if isinstance(zero_val, bool):
            zero_val = 'y' if zero_val else 'n'
        args.extend(['-Z', zero_val])
    if 'readahead' in options:
        readahead_val = options.pop('readahead')
        # Convert numeric to string if needed
        args.extend(['-r', str(readahead_val)])

    result = self._run('lvconvert', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

create(*args, **options)

Create Logical Volume.

Creates a new LV in the specified VG: - Allocates space from VG - Creates device mapper device - Initializes LV metadata

Parameters:

Name Type Description Default
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.create(size='1G')
True
Source code in sts_libs/src/sts/lvm.py 
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
def create(self, *args: str, **options: str) -> bool:
    """Create Logical Volume.

    Creates a new LV in the specified VG:
    - Allocates space from VG
    - Creates device mapper device
    - Initializes LV metadata

    Args:
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.create(size='1G')
        True
        ```
    """
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.vg:
        logging.error('Volume group required')
        return False

    result = self._run('lvcreate', '-n', self.name, self.vg, *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

create_snapshot(snapshot_name, *args, **options)

Create snapshot of this LV.

Creates a snapshot of the current LV (self) with the given snapshot name. For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

Parameters:

Name Type Description Default
snapshot_name str

Name for the new snapshot LV

 required
*args str

Additional command-line arguments (e.g., '-K', '-k', '--test')

 ()
**options str

Snapshot options (see LvmOptions)

 {}

Returns:

Type Description
LogicalVolume | None

LogicalVolume instance representing the created snapshot, or None if failed
Example 
# Create origin LV and then snapshot it
origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
# ... create the origin LV ...
# Create thin snapshot (no size needed)
snap1 = origin_lv.create_snapshot('snap1')

# Create COW snapshot with ignore activation skip
snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')
Source code in sts_libs/src/sts/lvm.py 
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
1728
1729
1730
1731
1732
1733
1734
def create_snapshot(self, snapshot_name: str, *args: str, **options: str) -> LogicalVolume | None:
    """Create snapshot of this LV.

    Creates a snapshot of the current LV (self) with the given snapshot name.
    For thin LVs, creates thin snapshots. For regular LVs, creates COW snapshots.

    Args:
        snapshot_name: Name for the new snapshot LV
        *args: Additional command-line arguments (e.g., '-K', '-k', '--test')
        **options: Snapshot options (see LvmOptions)

    Returns:
        LogicalVolume instance representing the created snapshot, or None if failed

    Example:
        ```python
        # Create origin LV and then snapshot it
        origin_lv = LogicalVolume(name='thin_lv', vg='vg0')
        # ... create the origin LV ...
        # Create thin snapshot (no size needed)
        snap1 = origin_lv.create_snapshot('snap1')

        # Create COW snapshot with ignore activation skip
        snap2 = origin_lv.create_snapshot('snap2', '-K', size='100M')
        ```
    """
    if not self.name:
        logging.error('Origin LV name required')
        return None
    if not self.vg:
        logging.error('Volume group required')
        return None
    if not snapshot_name:
        logging.error('Snapshot name required')
        return None

    # Build snapshot command
    cmd_args = ['-s']

    # Add any additional arguments
    if args:
        cmd_args.extend(args)

    # Add origin (this LV)
    cmd_args.append(f'{self.vg}/{self.name}')

    # Add snapshot name
    cmd_args.extend(['-n', snapshot_name])

    result = self._run('lvcreate', *cmd_args, **options)
    if result.succeeded:
        self.refresh_report()
        # Return new LogicalVolume instance representing the snapshot
        snap = LogicalVolume(name=snapshot_name, vg=self.vg)
        snap.refresh_report()
        return snap
    return None

create_thin_pool(pool_name, vg_name, **options) classmethod

Create thin pool with specified options.

Parameters:

Name Type Description Default
pool_name str

Pool name

 required
vg_name str

Volume group name

 required
**options str

Pool creation options

 {}

Returns:

Type Description
LogicalVolume

LogicalVolume object for the created pool

Raises:

Type Description
AssertionError

If pool creation fails
Source code in sts_libs/src/sts/lvm.py 
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
@classmethod
def create_thin_pool(cls, pool_name: str, vg_name: str, **options: str) -> LogicalVolume:
    """Create thin pool with specified options.

    Args:
        pool_name: Pool name
        vg_name: Volume group name
        **options: Pool creation options

    Returns:
        LogicalVolume object for the created pool

    Raises:
        AssertionError: If pool creation fails
    """
    pool = cls(name=pool_name, vg=vg_name)
    options['type'] = 'thin-pool'
    assert pool.create(**options), f'Failed to create thin pool {pool_name}'
    return pool

create_thin_volume(lv_name, vg_name, pool_name, virtualsize, **options) classmethod

Create thin volume in specified pool.

Parameters:

Name Type Description Default
lv_name str

Thin volume name

 required
vg_name str

Volume group name

 required
pool_name str

Parent pool name

 required
virtualsize str

Virtual size for thin volume

 required
**options str

Thin volume creation options

 {}

Returns:

Type Description
LogicalVolume

LogicalVolume object for the created thin volume

Raises:

Type Description
AssertionError

If volume creation fails
Source code in sts_libs/src/sts/lvm.py 
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
@classmethod
def create_thin_volume(
    cls, lv_name: str, vg_name: str, pool_name: str, virtualsize: str, **options: str
) -> LogicalVolume:
    """Create thin volume in specified pool.

    Args:
        lv_name: Thin volume name
        vg_name: Volume group name
        pool_name: Parent pool name
        virtualsize: Virtual size for thin volume
        **options: Thin volume creation options

    Returns:
        LogicalVolume object for the created thin volume

    Raises:
        AssertionError: If volume creation fails
    """
    thin_lv = cls(name=lv_name, pool_name=pool_name, vg=vg_name)
    assert thin_lv.create(virtualsize=virtualsize, type='thin', thinpool=pool_name, **options), (
        f'Failed to create thin volume {lv_name}'
    )
    return thin_lv

deactivate()

Deactivate Logical Volume.

Source code in sts_libs/src/sts/lvm.py 
1623
1624
1625
1626
1627
1628
1629
1630
def deactivate(self) -> bool:
    """Deactivate Logical Volume."""
    udevadm_settle()
    result = self.change('-an', f'{self.vg}/{self.name}')
    udevadm_settle()
    if result:
        return self.wait_for_lv_deactivation()
    return result

discover_vg()

Discover VG if name is available.

Source code in sts_libs/src/sts/lvm.py 
1005
1006
1007
1008
1009
1010
1011
1012
def discover_vg(self) -> str | None:
    """Discover VG if name is available."""
    if self.name and not self.vg:
        result = run(f'lvs {self.name} -o vg_name --noheadings')
        if result.succeeded:
            self.vg = result.stdout.strip()
            return self.vg
    return None

display(**options)

Display Logical Volume details.

Shows detailed information about the LV: - Size and allocation - Attributes and permissions - Segment information - Device mapper details

Parameters:

Name Type Description Default
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
CommandResult

CommandResult object containing command output and status
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
result = lv.display()
print(result.stdout)
Source code in sts_libs/src/sts/lvm.py 
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
def display(self, **options: str) -> CommandResult:
    """Display Logical Volume details.

    Shows detailed information about the LV:
    - Size and allocation
    - Attributes and permissions
    - Segment information
    - Device mapper details

    Args:
        **options: LV options (see LvmOptions)

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        result = lv.display()
        print(result.stdout)
        ```
    """
    if not self.vg or not self.name:
        return self._run('lvdisplay', **options)
    return self._run('lvdisplay', f'{self.vg}/{self.name}', **options)

extend(**options)

Extend Logical volume.

  • LV must be initialized (using lvcreate)
  • VG must have sufficient usable space

Parameters:

Name Type Description Default
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lvol0', vg='vg0')
lv.extend(extents='100%vg')
True
Source code in sts_libs/src/sts/lvm.py 
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
def extend(self, **options: str) -> bool:
    """Extend Logical volume.

    - LV must be initialized (using lvcreate)
    - VG must have sufficient usable space

    Args:
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lvol0', vg='vg0')
        lv.extend(extents='100%vg')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvextend', f'{self.vg}/{self.name}', **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

from_report(report) classmethod

Create LogicalVolume from LVReport.

Parameters:

Name Type Description Default
report LVReport

LV report data

 required

Returns:

Type Description
LogicalVolume | None

LogicalVolume instance or None if invalid
Example 
lv = LogicalVolume.from_report(report)
Source code in sts_libs/src/sts/lvm.py 
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
@classmethod
def from_report(cls, report: LVReport) -> LogicalVolume | None:
    """Create LogicalVolume from LVReport.

    Args:
        report: LV report data

    Returns:
        LogicalVolume instance or None if invalid

    Example:
        ```python
        lv = LogicalVolume.from_report(report)
        ```
    """
    if not report.name or not report.vg:
        return None

    # Create LogicalVolume with report already attached
    return cls(
        name=report.name,
        vg=report.vg,
        path=report.lv_path,
        report=report,  # Attach the report directly
        prevent_report_updates=True,  # Avoid double refresh since report is already fresh
    )

get_all(vg=None) classmethod

Get all Logical Volumes.

Parameters:

Name Type Description Default
vg str | None

Optional volume group to filter by

 None

Returns:

Type Description
list[LogicalVolume]

List of LogicalVolume instances
Example 
LogicalVolume.get_all()
[LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]
Source code in sts_libs/src/sts/lvm.py 
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
@classmethod
def get_all(cls, vg: str | None = None) -> list[LogicalVolume]:
    """Get all Logical Volumes.

    Args:
        vg: Optional volume group to filter by

    Returns:
        List of LogicalVolume instances

    Example:
        ```python
        LogicalVolume.get_all()
        [LogicalVolume(name='lv0', vg='vg0', ...), LogicalVolume(name='lv1', vg='vg1', ...)]
        ```
    """
    logical_volumes: list[LogicalVolume] = []

    # Get all reports
    reports = LVReport.get_all(vg)

    # Create LogicalVolumes from reports
    logical_volumes.extend(lv for report in reports if (lv := cls.from_report(report)))

    return logical_volumes

get_data_stripe_size()

Get stripe size for thin pool data component.

For thin pools, the actual stripe size information is stored in the data component (_tdata), not in the main pool LV. This method returns the stripe size from the data component.

Returns:

Type Description
str | None

String representing stripe size, or None if not a thin pool or stripe size not found
Example 
pool = LogicalVolume(name='pool', vg='vg0')
pool.create(stripes='2', stripesize='64k', size='100M', type='thin-pool')
stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'
Source code in sts_libs/src/sts/lvm.py 
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
def get_data_stripe_size(self) -> str | None:
    """Get stripe size for thin pool data component.

    For thin pools, the actual stripe size information is stored in the data component (_tdata),
    not in the main pool LV. This method returns the stripe size from the data component.

    Returns:
        String representing stripe size, or None if not a thin pool or stripe size not found

    Example:
        ```python
        pool = LogicalVolume(name='pool', vg='vg0')
        pool.create(stripes='2', stripesize='64k', size='100M', type='thin-pool')
        stripe_size = pool.get_data_stripe_size()  # Returns '64.00k'
        ```
    """
    if not self.report or not self.report.data_lv:
        # Not a thin pool or no data component
        return None

    # Get the data component name (usually pool_name + '_tdata')
    data_lv_name = self.report.data_lv
    if not data_lv_name:
        return None

    # Remove brackets if present (e.g., [pool1_tdata] -> pool1_tdata)
    data_lv_name = data_lv_name.strip('[]')

    # Create LogicalVolume for the data component and get its stripe size
    try:
        data_lv = LogicalVolume(name=data_lv_name, vg=self.vg)
        if data_lv.refresh_report() and data_lv.report:
            return data_lv.report.stripe_size
    except (ValueError, OSError) as e:
        logging.warning(f'Failed to get stripe size info from data component {data_lv_name}: {e}')

    return None

get_data_stripes()

Get stripe count for thin pool data component.

For thin pools, the actual stripe information is stored in the data component (_tdata), not in the main pool LV. This method returns the stripe count from the data component.

Returns:

Type Description
str | None

String representing stripe count, or None if not a thin pool or stripes not found
Example 
pool = LogicalVolume(name='pool', vg='vg0')
pool.create(stripes='2', size='100M', type='thin-pool')
stripe_count = pool.get_data_stripes()  # Returns '2'
Source code in sts_libs/src/sts/lvm.py 
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
def get_data_stripes(self) -> str | None:
    """Get stripe count for thin pool data component.

    For thin pools, the actual stripe information is stored in the data component (_tdata),
    not in the main pool LV. This method returns the stripe count from the data component.

    Returns:
        String representing stripe count, or None if not a thin pool or stripes not found

    Example:
        ```python
        pool = LogicalVolume(name='pool', vg='vg0')
        pool.create(stripes='2', size='100M', type='thin-pool')
        stripe_count = pool.get_data_stripes()  # Returns '2'
        ```
    """
    if not self.report or not self.report.data_lv:
        # Not a thin pool or no data component
        return None

    # Get the data component name (usually pool_name + '_tdata')
    data_lv_name = self.report.data_lv
    if not data_lv_name:
        return None

    # Remove brackets if present (e.g., [pool1_tdata] -> pool1_tdata)
    data_lv_name = data_lv_name.strip('[]')

    # Create LogicalVolume for the data component and get its stripes
    try:
        data_lv = LogicalVolume(name=data_lv_name, vg=self.vg)
        if data_lv.refresh_report() and data_lv.report:
            return data_lv.report.stripes
    except (ValueError, OSError) as e:
        logging.warning(f'Failed to get stripe info from data component {data_lv_name}: {e}')

    return None

get_pool_usage()

Get thin pool data and metadata usage percentages.

Returns:

Type Description
tuple[str, str]

Tuple of (data_percent, metadata_percent) as strings
Source code in sts_libs/src/sts/lvm.py 
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
def get_pool_usage(self) -> tuple[str, str]:
    """Get thin pool data and metadata usage percentages.

    Returns:
        Tuple of (data_percent, metadata_percent) as strings
    """
    if not self.vg or not self.name:
        logging.error('Volume group and logical volume name required')
        return 'unknown', 'unknown'

    # Get data usage
    data_result = self.lvs(f'{self.vg}/{self.name}', o='data_percent', noheadings='')
    data_percent = data_result.stdout.strip() if data_result.succeeded else 'unknown'

    # Get metadata usage
    meta_result = self.lvs(f'{self.vg}/{self.name}', o='metadata_percent', noheadings='')
    meta_percent = meta_result.stdout.strip() if meta_result.succeeded else 'unknown'

    return data_percent, meta_percent

lvs(*args, **options)

Get information about logical volumes.

Executes the 'lvs' command with optional filtering to display information about logical volumes.

Parameters:

Name Type Description Default
*args str

Positional args passed through to lvs (e.g., LV selector, flags).

 ()
**options str

LV command options (see LvmOptions).

 {}

Returns:

Type Description
CommandResult

CommandResult object containing command output and status
Example 
lv = LogicalVolume()
result = lv.lvs()
print(result.stdout)
Source code in sts_libs/src/sts/lvm.py 
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
def lvs(self, *args: str, **options: str) -> CommandResult:
    """Get information about logical volumes.

    Executes the 'lvs' command with optional filtering to display
    information about logical volumes.

    Args:
        *args: Positional args passed through to `lvs` (e.g., LV selector, flags).
        **options: LV command options (see LvmOptions).

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume()
        result = lv.lvs()
        print(result.stdout)
        ```
    """
    return self._run('lvs', *args, **options)

reduce(*args, **options)

Reduce Logical Volume size.

Reduces LV size (shrinks the volume): - Filesystem must be shrunk first - Data loss risk if not done carefully - Cannot reduce below used space

Parameters:

Name Type Description Default
*args str

Additional lvreduce arguments

 ()
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.reduce(size='500M')
True

# With additional arguments
lv.reduce('--test', size='500M')
True
Source code in sts_libs/src/sts/lvm.py 
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
def reduce(self, *args: str, **options: str) -> bool:
    """Reduce Logical Volume size.

    Reduces LV size (shrinks the volume):
    - Filesystem must be shrunk first
    - Data loss risk if not done carefully
    - Cannot reduce below used space

    Args:
        *args: Additional lvreduce arguments
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.reduce(size='500M')
        True

        # With additional arguments
        lv.reduce('--test', size='500M')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    result = self._run('lvreduce', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

refresh_report()

Refresh LV report data.

Creates or updates the LV report with the latest information.

Returns:

Name Type Description
bool bool

True if refresh was successful
Source code in sts_libs/src/sts/lvm.py 
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
def refresh_report(self) -> bool:
    """Refresh LV report data.

    Creates or updates the LV report with the latest information.

    Returns:
        bool: True if refresh was successful
    """
    # Create new report if needed
    if not self.report:
        # Do not provide name and vg during init to prevent update
        self.report = LVReport()
        self.report.name = self.name
        self.report.vg = self.vg

    # Refresh the report data
    return self.report.refresh()

remove(*args, **options)

Remove Logical Volume.

Removes LV and its data: - Data is permanently lost - Space is returned to VG - Device mapper device is removed

Parameters:

Name Type Description Default
*args str

Additional volume paths to remove (for removing multiple volumes)

 ()
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
# Remove single volume
lv = LogicalVolume(name='lv0', vg='vg0')
lv.remove()
True

# Remove multiple volumes
lv = LogicalVolume(name='lv1', vg='vg0')
lv.remove('vg0/lv2', 'vg0/lv3')
True
Source code in sts_libs/src/sts/lvm.py 
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
def remove(self, *args: str, **options: str) -> bool:
    """Remove Logical Volume.

    Removes LV and its data:
    - Data is permanently lost
    - Space is returned to VG
    - Device mapper device is removed

    Args:
        *args: Additional volume paths to remove (for removing multiple volumes)
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        # Remove single volume
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.remove()
        True

        # Remove multiple volumes
        lv = LogicalVolume(name='lv1', vg='vg0')
        lv.remove('vg0/lv2', 'vg0/lv3')
        True
        ```
    """
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not self.vg:
        logging.error('Volume group required')
        return False

    # Start with this LV
    targets = [f'{self.vg}/{self.name}']

    # Add any additional volumes from args
    if args:
        targets.extend(args)

    result = self._run('lvremove', *targets, **options)
    return result.succeeded

rename(new_name, **options)

Rename Logical Volume.

Changes the LV name: - Must not conflict with existing LV names - Updates device mapper devices - May require remounting if mounted

Parameters:

Name Type Description Default
new_name str

New name for the LV

 required
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.rename('new_lv')
True
Source code in sts_libs/src/sts/lvm.py 
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
def rename(self, new_name: str, **options: str) -> bool:
    """Rename Logical Volume.

    Changes the LV name:
    - Must not conflict with existing LV names
    - Updates device mapper devices
    - May require remounting if mounted

    Args:
        new_name: New name for the LV
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.rename('new_lv')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False
    if not new_name:
        logging.error('New name required')
        return False

    result = self._run('lvrename', f'{self.vg}/{self.name}', new_name, **options)
    if result.succeeded:
        self.name = new_name
        self.path = f'/dev/{self.vg}/{self.name}'
        self.refresh_report()
    return result.succeeded

resize(*args, **options)

Resize Logical Volume.

Changes LV size (can grow or shrink): - Combines extend and reduce functionality - Safer than lvreduce for shrinking - Can resize filesystem simultaneously

Parameters:

Name Type Description Default
*args str

Additional lvresize arguments (e.g., '-l+2', '-t', '--test')

 ()
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
lv = LogicalVolume(name='lv0', vg='vg0')
lv.resize(size='2G')
True

# With additional arguments
lv.resize('-l+2', size='2G')
True

# With test flag
lv.resize('--test', size='2G')
True
Source code in sts_libs/src/sts/lvm.py 
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
def resize(self, *args: str, **options: str) -> bool:
    """Resize Logical Volume.

    Changes LV size (can grow or shrink):
    - Combines extend and reduce functionality
    - Safer than lvreduce for shrinking
    - Can resize filesystem simultaneously

    Args:
        *args: Additional lvresize arguments (e.g., '-l+2', '-t', '--test')
        **options: LV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        lv = LogicalVolume(name='lv0', vg='vg0')
        lv.resize(size='2G')
        True

        # With additional arguments
        lv.resize('-l+2', size='2G')
        True

        # With test flag
        lv.resize('--test', size='2G')
        True
        ```
    """
    if not self.vg:
        logging.error('Volume group name required')
        return False
    if not self.name:
        logging.error('Logical volume name required')
        return False

    result = self._run('lvresize', f'{self.vg}/{self.name}', *args, **options)
    if result.succeeded:
        self.refresh_report()
    return result.succeeded

scan(*args, **options)

Scan for Logical Volumes.

Scans all devices for LV information: - Discovers new LVs - Updates device mapper - Useful after system changes

Parameters:

Name Type Description Default
**options str

LV options (see LvmOptions)

 {}

Returns:

Type Description
CommandResult

CommandResult object containing command output and status
Example 
lv = LogicalVolume()
result = lv.scan()
print(result.stdout)
Source code in sts_libs/src/sts/lvm.py 
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
def scan(self, *args: str, **options: str) -> CommandResult:
    """Scan for Logical Volumes.

    Scans all devices for LV information:
    - Discovers new LVs
    - Updates device mapper
    - Useful after system changes

    Args:
        **options: LV options (see LvmOptions)

    Returns:
        CommandResult object containing command output and status

    Example:
        ```python
        lv = LogicalVolume()
        result = lv.scan()
        print(result.stdout)
        ```
    """
    return self._run('lvscan', *args, **options)

wait_for_lv_deactivation(timeout=30)

Wait for logical volume to be fully deactivated.

Parameters:

Name Type Description Default
timeout int

Maximum wait time in seconds

 30

Returns:

Type Description
bool

True if deactivated successfully, False if timeout
Source code in sts_libs/src/sts/lvm.py 
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
def wait_for_lv_deactivation(self, timeout: int = 30) -> bool:
    """Wait for logical volume to be fully deactivated.

    Args:
        timeout: Maximum wait time in seconds

    Returns:
        True if deactivated successfully, False if timeout
    """
    start_time = time.time()

    while time.time() - start_time < timeout:
        # Check LV status using lvs command
        result = self.lvs(f'{self.vg}/{self.name}', '--noheadings', '-o lv_active')
        if result.succeeded and 'active' not in result.stdout.lower():
            # LV is inactive - also verify device node is gone
            if self.path is not None:
                device_path = Path(self.path)
                if not device_path.exists():
                    return True
            else:
                return True  # If no path, consider it deactivated
        time.sleep(2)  # Poll every 2 seconds

    logging.warning(f'LV {self.vg}/{self.name} deactivation timed out after {timeout}s')
    return False

LvmDevice dataclass

Bases: StorageDevice

Base class for LVM devices.

Provides common functionality for all LVM device types: - Command execution with standard options - Configuration management - Basic device operations

Parameters:

Name Type Description Default
name str | None

Device name (optional)

 None
path Path | str | None

Device path (optional, defaults to /dev/)

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
model str | None

Device model (optional)

 None
yes bool

Automatically answer yes to prompts

 True
force bool

Force operations without confirmation

 False

The yes and force options are useful for automation: - yes: Skip interactive prompts - force: Ignore warnings and errors

Source code in sts_libs/src/sts/lvm.py 
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
558
559
560
561
562
563
564
565
@dataclass
class LvmDevice(StorageDevice):
    """Base class for LVM devices.

    Provides common functionality for all LVM device types:
    - Command execution with standard options
    - Configuration management
    - Basic device operations

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation

    The yes and force options are useful for automation:
    - yes: Skip interactive prompts
    - force: Ignore warnings and errors
    """

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

    # Optional parameters for this class
    yes: bool = True  # Answer yes to prompts
    force: bool = False  # Force operations

    # Internal fields
    _config_path: Path = field(init=False, default=Path('/etc/lvm/lvm.conf'))

    def __post_init__(self) -> None:
        """Initialize LVM device."""
        # Set path based on name if not provided
        if not self.path and self.name:
            self.path = f'/dev/{self.name}'

        # Initialize parent class
        super().__post_init__()

    def _run(self, cmd: str, *args: str | Path | None, **kwargs: str) -> CommandResult:
        """Run LVM command.

        Builds and executes LVM commands with standard options:
        - Adds --yes for non-interactive mode
        - Adds --force to ignore warnings
        - Converts Python parameters to LVM options

        Args:
            cmd: Command name (e.g. 'pvcreate')
            *args: Command arguments
            **kwargs: Command parameters

        Returns:
            Command result
        """
        command = [cmd]
        if self.yes:
            command.append('--yes')
        if self.force:
            command.append('--force')
        if args:
            command.extend(str(arg) for arg in args if arg)
        if kwargs:
            command.extend(f'--{k.replace("_", "-")}={v}' for k, v in kwargs.items() if v)

        return run(' '.join(command))

    @abstractmethod
    def create(self, **options: str) -> bool:
        """Create LVM device.

        Args:
            **options: Device options (see LvmOptions)

        Returns:
            True if successful, False otherwise
        """

    @abstractmethod
    def remove(self, **options: str) -> bool:
        """Remove LVM device.

        Args:
            **options: Device options (see LvmOptions)

        Returns:
            True if successful, False otherwise
        """

__post_init__()

Initialize LVM device.

Source code in sts_libs/src/sts/lvm.py 
508
509
510
511
512
513
514
515
def __post_init__(self) -> None:
    """Initialize LVM device."""
    # Set path based on name if not provided
    if not self.path and self.name:
        self.path = f'/dev/{self.name}'

    # Initialize parent class
    super().__post_init__()

create(**options) abstractmethod

Create LVM device.

Parameters:

Name Type Description Default
**options str

Device options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Source code in sts_libs/src/sts/lvm.py 
545
546
547
548
549
550
551
552
553
554
@abstractmethod
def create(self, **options: str) -> bool:
    """Create LVM device.

    Args:
        **options: Device options (see LvmOptions)

    Returns:
        True if successful, False otherwise
    """

remove(**options) abstractmethod

Remove LVM device.

Parameters:

Name Type Description Default
**options str

Device options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Source code in sts_libs/src/sts/lvm.py 
556
557
558
559
560
561
562
563
564
565
@abstractmethod
def remove(self, **options: str) -> bool:
    """Remove LVM device.

    Args:
        **options: Device options (see LvmOptions)

    Returns:
        True if successful, False otherwise
    """

LvmOptions

Bases: TypedDict

LVM command options.

Common options: - size: Volume size (e.g. '1G', '500M') - extents: Volume size in extents (e.g. '100%FREE') - permission: Volume permission (rw/r) - persistent: Make settings persistent across reboots - monitor: Monitor volume for events - autobackup: Auto backup metadata after changes

Size can be specified in: - Absolute size (1G, 500M) - Percentage of VG (80%VG) - Percentage of free space (100%FREE) - Physical extents (100)

Source code in sts_libs/src/sts/lvm.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
class LvmOptions(TypedDict, total=False):
    """LVM command options.

    Common options:
    - size: Volume size (e.g. '1G', '500M')
    - extents: Volume size in extents (e.g. '100%FREE')
    - permission: Volume permission (rw/r)
    - persistent: Make settings persistent across reboots
    - monitor: Monitor volume for events
    - autobackup: Auto backup metadata after changes

    Size can be specified in:
    - Absolute size (1G, 500M)
    - Percentage of VG (80%VG)
    - Percentage of free space (100%FREE)
    - Physical extents (100)
    """

    size: str
    extents: str
    permission: str
    persistent: str
    monitor: str
    autobackup: str

PVInfo dataclass

Physical Volume information.

Stores key information about a Physical Volume: - Volume group membership - Format type (lvm2) - Attributes (allocatable, exported, etc) - Size information (total and free space)

Parameters:

Name Type Description Default
vg str | None

Volume group name (None if not in a VG)

 required
fmt str

PV format (usually 'lvm2')

 required
attr str

PV attributes (e.g. 'a--' for allocatable)

 required
psize str

PV size (e.g. '1.00t')

 required
pfree str

PV free space (e.g. '500.00g')

 required
Source code in sts_libs/src/sts/lvm.py 
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
@dataclass
class PVInfo:
    """Physical Volume information.

    Stores key information about a Physical Volume:
    - Volume group membership
    - Format type (lvm2)
    - Attributes (allocatable, exported, etc)
    - Size information (total and free space)

    Args:
        vg: Volume group name (None if not in a VG)
        fmt: PV format (usually 'lvm2')
        attr: PV attributes (e.g. 'a--' for allocatable)
        psize: PV size (e.g. '1.00t')
        pfree: PV free space (e.g. '500.00g')
    """

    vg: str | None
    fmt: str
    attr: str
    psize: str
    pfree: str

PhysicalVolume dataclass

Bases: LvmDevice

Physical Volume device.

A Physical Volume (PV) is a disk or partition used by LVM. PVs provide the storage pool for Volume Groups.

Key features: - Initialize disks/partitions for LVM use - Track space allocation - Handle bad block management - Store LVM metadata

Parameters:

Name Type Description Default
name str | None

Device name (optional)

 None
path Path | str | None

Device path (optional, defaults to /dev/)

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
model str | None

Device model (optional)

 None
yes bool

Automatically answer yes to prompts

 True
force bool

Force operations without confirmation

 False
vg str | None

Volume group name (optional, discovered from device)

 None
fmt str | None

PV format (optional, discovered from device)

 None
attr str | None

PV attributes (optional, discovered from device)

 None
pfree str | None

PV free space (optional, discovered from device)

 None
Example 
pv = PhysicalVolume(name='sda1')  # Discovers other values
pv = PhysicalVolume.create('/dev/sda1')  # Creates new PV
Source code in sts_libs/src/sts/lvm.py 
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
@dataclass
class PhysicalVolume(LvmDevice):
    """Physical Volume device.

    A Physical Volume (PV) is a disk or partition used by LVM.
    PVs provide the storage pool for Volume Groups.

    Key features:
    - Initialize disks/partitions for LVM use
    - Track space allocation
    - Handle bad block management
    - Store LVM metadata

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        vg: Volume group name (optional, discovered from device)
        fmt: PV format (optional, discovered from device)
        attr: PV attributes (optional, discovered from device)
        pfree: PV free space (optional, discovered from device)

    Example:
        ```python
        pv = PhysicalVolume(name='sda1')  # Discovers other values
        pv = PhysicalVolume.create('/dev/sda1')  # Creates new PV
        ```
    """

    # Optional parameters for this class
    vg: str | None = None  # Volume Group membership
    fmt: str | None = None  # PV format (usually lvm2)
    attr: str | None = None  # PV attributes
    pfree: str | None = None  # Free space

    # Available PV commands
    COMMANDS: ClassVar[list[str]] = [
        'pvchange',  # Modify PV attributes
        'pvck',  # Check PV metadata
        'pvcreate',  # Initialize PV
        'pvdisplay',  # Show PV details
        'pvmove',  # Move PV data
        'pvremove',  # Remove PV
        'pvresize',  # Resize PV
        'pvs',  # List PVs
        'pvscan',  # Scan for PVs
    ]

    # Discover PV info if path is available
    def discover_pv_info(self) -> None:
        """Discovers PV information if path is available.

        Volume group membership.
        Format and attributes.
        Size information.
        """
        result = run(f'pvs {self.path} --noheadings --separator ","')
        if result.succeeded:
            # Parse PV info line
            # Format: PV,VG,Fmt,Attr,PSize,PFree
            parts = result.stdout.strip().split(',')
            if len(parts) == 6:
                _, vg, fmt, attr, _, pfree = parts
                if not self.vg:
                    self.vg = vg or None
                if not self.fmt:
                    self.fmt = fmt
                if not self.attr:
                    self.attr = attr
                if not self.pfree:
                    self.pfree = pfree

    def create(self, **options: str) -> bool:
        """Create Physical Volume.

        Initializes a disk or partition for use with LVM:
        - Creates LVM metadata area
        - Prepares device for VG membership

        Args:
            **options: PV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            pv = PhysicalVolume(path='/dev/sda1')
            pv.create()
            True
            ```
        """
        if not self.path:
            logging.error('Device path required')
            return False

        result = self._run('pvcreate', str(self.path), **options)
        if result.succeeded:
            self.discover_pv_info()
        return result.succeeded

    def remove(self, **options: str) -> bool:
        """Remove Physical Volume.

        Removes LVM metadata from device:
        - Device must not be in use by a VG
        - Data on device is not erased

        Args:
            **options: PV options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            pv = PhysicalVolume(path='/dev/sda1')
            pv.remove()
            True
            ```
        """
        if not self.path:
            logging.error('Device path required')
            return False

        result = self._run('pvremove', str(self.path), **options)
        return result.succeeded

    @classmethod
    def get_all(cls) -> dict[str, PVInfo]:
        """Get all Physical Volumes.

        Returns:
            Dictionary mapping PV names to their information

        Example:
            ```python
            PhysicalVolume.get_all()
            {'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}
            ```
        """
        result = run('pvs --noheadings --separator ","')
        if result.failed:
            logging.debug('No Physical Volumes found')
            return {}

        # Format: PV,VG,Fmt,Attr,PSize,PFree
        pv_info_regex = r'\s+(\S+),(\S+)?,(\S+),(.*),(.*),(.*)$'
        pv_dict = {}

        for line in result.stdout.splitlines():
            if match := re.match(pv_info_regex, line):
                pv_dict[match.group(1)] = PVInfo(
                    vg=match.group(2) or None,  # VG can be empty
                    fmt=match.group(3),
                    attr=match.group(4),
                    psize=match.group(5),
                    pfree=match.group(6),
                )

        return pv_dict

create(**options)

Create Physical Volume.

Initializes a disk or partition for use with LVM: - Creates LVM metadata area - Prepares device for VG membership

Parameters:

Name Type Description Default
**options str

PV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
pv = PhysicalVolume(path='/dev/sda1')
pv.create()
True
Source code in sts_libs/src/sts/lvm.py 
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
def create(self, **options: str) -> bool:
    """Create Physical Volume.

    Initializes a disk or partition for use with LVM:
    - Creates LVM metadata area
    - Prepares device for VG membership

    Args:
        **options: PV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        pv = PhysicalVolume(path='/dev/sda1')
        pv.create()
        True
        ```
    """
    if not self.path:
        logging.error('Device path required')
        return False

    result = self._run('pvcreate', str(self.path), **options)
    if result.succeeded:
        self.discover_pv_info()
    return result.succeeded

discover_pv_info()

Discovers PV information if path is available.

Volume group membership. Format and attributes. Size information.

Source code in sts_libs/src/sts/lvm.py 
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
def discover_pv_info(self) -> None:
    """Discovers PV information if path is available.

    Volume group membership.
    Format and attributes.
    Size information.
    """
    result = run(f'pvs {self.path} --noheadings --separator ","')
    if result.succeeded:
        # Parse PV info line
        # Format: PV,VG,Fmt,Attr,PSize,PFree
        parts = result.stdout.strip().split(',')
        if len(parts) == 6:
            _, vg, fmt, attr, _, pfree = parts
            if not self.vg:
                self.vg = vg or None
            if not self.fmt:
                self.fmt = fmt
            if not self.attr:
                self.attr = attr
            if not self.pfree:
                self.pfree = pfree

get_all() classmethod

Get all Physical Volumes.

Returns:

Type Description
dict[str, PVInfo]

Dictionary mapping PV names to their information
Example 
PhysicalVolume.get_all()
{'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}
Source code in sts_libs/src/sts/lvm.py 
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
@classmethod
def get_all(cls) -> dict[str, PVInfo]:
    """Get all Physical Volumes.

    Returns:
        Dictionary mapping PV names to their information

    Example:
        ```python
        PhysicalVolume.get_all()
        {'/dev/sda1': PVInfo(vg='vg0', fmt='lvm2', ...)}
        ```
    """
    result = run('pvs --noheadings --separator ","')
    if result.failed:
        logging.debug('No Physical Volumes found')
        return {}

    # Format: PV,VG,Fmt,Attr,PSize,PFree
    pv_info_regex = r'\s+(\S+),(\S+)?,(\S+),(.*),(.*),(.*)$'
    pv_dict = {}

    for line in result.stdout.splitlines():
        if match := re.match(pv_info_regex, line):
            pv_dict[match.group(1)] = PVInfo(
                vg=match.group(2) or None,  # VG can be empty
                fmt=match.group(3),
                attr=match.group(4),
                psize=match.group(5),
                pfree=match.group(6),
            )

    return pv_dict

remove(**options)

Remove Physical Volume.

Removes LVM metadata from device: - Device must not be in use by a VG - Data on device is not erased

Parameters:

Name Type Description Default
**options str

PV options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
pv = PhysicalVolume(path='/dev/sda1')
pv.remove()
True
Source code in sts_libs/src/sts/lvm.py 
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
def remove(self, **options: str) -> bool:
    """Remove Physical Volume.

    Removes LVM metadata from device:
    - Device must not be in use by a VG
    - Data on device is not erased

    Args:
        **options: PV options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        pv = PhysicalVolume(path='/dev/sda1')
        pv.remove()
        True
        ```
    """
    if not self.path:
        logging.error('Device path required')
        return False

    result = self._run('pvremove', str(self.path), **options)
    return result.succeeded

VolumeGroup dataclass

Bases: LvmDevice

Volume Group device.

A Volume Group (VG) combines Physical Volumes into a storage pool. This pool can then be divided into Logical Volumes.

Key features: - Combine multiple PVs - Manage storage pool - Track extent allocation - Handle PV addition/removal

Parameters:

Name Type Description Default
name str | None

Device name (optional)

 None
path Path | str | None

Device path (optional, defaults to /dev/)

 None
size int | None

Device size in bytes (optional, discovered from device)

 None
model str | None

Device model (optional)

 None
yes bool

Automatically answer yes to prompts

 True
force bool

Force operations without confirmation

 False
pvs list[str]

List of Physical Volumes (optional, discovered from device)

 list()
Example 
vg = VolumeGroup(name='vg0')  # Discovers other values
vg = VolumeGroup.create('vg0', ['/dev/sda1'])  # Creates new VG
Source code in sts_libs/src/sts/lvm.py 
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
@dataclass
class VolumeGroup(LvmDevice):
    """Volume Group device.

    A Volume Group (VG) combines Physical Volumes into a storage pool.
    This pool can then be divided into Logical Volumes.

    Key features:
    - Combine multiple PVs
    - Manage storage pool
    - Track extent allocation
    - Handle PV addition/removal

    Args:
        name: Device name (optional)
        path: Device path (optional, defaults to /dev/<name>)
        size: Device size in bytes (optional, discovered from device)
        model: Device model (optional)
        yes: Automatically answer yes to prompts
        force: Force operations without confirmation
        pvs: List of Physical Volumes (optional, discovered from device)

    Example:
        ```python
        vg = VolumeGroup(name='vg0')  # Discovers other values
        vg = VolumeGroup.create('vg0', ['/dev/sda1'])  # Creates new VG
        ```
    """

    # Optional parameters for this class
    pvs: list[str] = field(default_factory=list)  # Member PVs

    # Available VG commands
    COMMANDS: ClassVar[list[str]] = [
        'vgcfgbackup',  # Backup VG metadata
        'vgcfgrestore',  # Restore VG metadata
        'vgchange',  # Change VG attributes
        'vgck',  # Check VG metadata
        'vgconvert',  # Convert VG metadata format
        'vgcreate',  # Create VG
        'vgdisplay',  # Show VG details
        'vgexport',  # Make VG inactive
        'vgextend',  # Add PVs to VG
        'vgimport',  # Make VG active
        'vgimportclone',  # Import cloned PVs
        'vgimportdevices',  # Import PVs into VG
        'vgmerge',  # Merge VGs
        'vgmknodes',  # Create VG special files
        'vgreduce',  # Remove PVs from VG
        'vgremove',  # Remove VG
        'vgrename',  # Rename VG
        'vgs',  # List VGs
        'vgscan',  # Scan for VGs
        'vgsplit',  # Split VG into two
    ]

    def discover_pvs(self) -> list[str] | None:
        """Discover PVs if name is available."""
        if self.name:
            result = run(f'vgs {self.name} -o pv_name --noheadings')
            if result.succeeded:
                self.pvs = result.stdout.strip().splitlines()
                return self.pvs
        return None

    def create(self, **options: str) -> bool:
        """Create Volume Group.

        Creates a new VG from specified PVs:
        - Initializes VG metadata
        - Sets up extent allocation
        - Creates device mapper devices

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
            vg.create()
            True
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False
        if not self.pvs:
            logging.error('Physical volumes required')
            return False

        result = self._run('vgcreate', self.name, *self.pvs, **options)
        return result.succeeded

    def remove(self, **options: str) -> bool:
        """Remove Volume Group.

        Removes VG and its metadata:
        - All LVs must be removed first
        - PVs are released but not removed

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.remove()
            True
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgremove', self.name, **options)
        return result.succeeded

    def activate(self, **options: str) -> bool:
        """Activate Volume Group.

        Makes the VG and all its LVs available for use.

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.activate()
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgchange', '-a', 'y', self.name, **options)
        return result.succeeded

    def deactivate(self, **options: str) -> bool:
        """Deactivate Volume Group.

        Makes the VG and all its LVs unavailable.

        Args:
            **options: VG options (see LvmOptions)

        Returns:
            True if successful, False otherwise

        Example:
            ```python
            vg = VolumeGroup(name='vg0')
            vg.deactivate()
            ```
        """
        if not self.name:
            logging.error('Volume group name required')
            return False

        result = self._run('vgchange', '-a', 'n', self.name, **options)
        return result.succeeded

activate(**options)

Activate Volume Group.

Makes the VG and all its LVs available for use.

Parameters:

Name Type Description Default
**options str

VG options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
vg = VolumeGroup(name='vg0')
vg.activate()
Source code in sts_libs/src/sts/lvm.py 
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
def activate(self, **options: str) -> bool:
    """Activate Volume Group.

    Makes the VG and all its LVs available for use.

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.activate()
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgchange', '-a', 'y', self.name, **options)
    return result.succeeded

create(**options)

Create Volume Group.

Creates a new VG from specified PVs: - Initializes VG metadata - Sets up extent allocation - Creates device mapper devices

Parameters:

Name Type Description Default
**options str

VG options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
vg.create()
True
Source code in sts_libs/src/sts/lvm.py 
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
def create(self, **options: str) -> bool:
    """Create Volume Group.

    Creates a new VG from specified PVs:
    - Initializes VG metadata
    - Sets up extent allocation
    - Creates device mapper devices

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0', pvs=['/dev/sda1'])
        vg.create()
        True
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False
    if not self.pvs:
        logging.error('Physical volumes required')
        return False

    result = self._run('vgcreate', self.name, *self.pvs, **options)
    return result.succeeded

deactivate(**options)

Deactivate Volume Group.

Makes the VG and all its LVs unavailable.

Parameters:

Name Type Description Default
**options str

VG options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
vg = VolumeGroup(name='vg0')
vg.deactivate()
Source code in sts_libs/src/sts/lvm.py 
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
def deactivate(self, **options: str) -> bool:
    """Deactivate Volume Group.

    Makes the VG and all its LVs unavailable.

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.deactivate()
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgchange', '-a', 'n', self.name, **options)
    return result.succeeded

discover_pvs()

Discover PVs if name is available.

Source code in sts_libs/src/sts/lvm.py 
790
791
792
793
794
795
796
797
def discover_pvs(self) -> list[str] | None:
    """Discover PVs if name is available."""
    if self.name:
        result = run(f'vgs {self.name} -o pv_name --noheadings')
        if result.succeeded:
            self.pvs = result.stdout.strip().splitlines()
            return self.pvs
    return None

remove(**options)

Remove Volume Group.

Removes VG and its metadata: - All LVs must be removed first - PVs are released but not removed

Parameters:

Name Type Description Default
**options str

VG options (see LvmOptions)

 {}

Returns:

Type Description
bool

True if successful, False otherwise
Example 
vg = VolumeGroup(name='vg0')
vg.remove()
True
Source code in sts_libs/src/sts/lvm.py 
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
def remove(self, **options: str) -> bool:
    """Remove Volume Group.

    Removes VG and its metadata:
    - All LVs must be removed first
    - PVs are released but not removed

    Args:
        **options: VG options (see LvmOptions)

    Returns:
        True if successful, False otherwise

    Example:
        ```python
        vg = VolumeGroup(name='vg0')
        vg.remove()
        True
        ```
    """
    if not self.name:
        logging.error('Volume group name required')
        return False

    result = self._run('vgremove', self.name, **options)
    return result.succeeded