community.general/plugins/modules/lvm_pv.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

# Copyright (c) 2025, Klention Mali <klention@gmail.com>
# Based on lvol module by Jeroen Hoekx <jeroen.hoekx@dsquare.be>
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import absolute_import, division, print_function
__metaclass__ = type


DOCUMENTATION = r"""
module: lvm_pv
short_description: Manage LVM Physical Volumes
version_added: "11.0.0"
description:
  - Creates, resizes or removes LVM Physical Volumes.
author:
  - Klention Mali (@klention)
options:
  device:
    description:
      - Path to the block device to manage.
    type: path
    required: true
  state:
    description:
      - Control if the physical volume exists.
    type: str
    choices: [present, absent]
    default: present
  force:
    description:
      - Force the operation.
      - When O(state=present) (creating a PV), this uses C(pvcreate -f) to force creation.
      - When O(state=absent) (removing a PV), this uses C(pvremove -ff) to force removal even if part of a volume group.
    type: bool
    default: false
  resize:
    description:
      - Resize PV to device size when O(state=present).
    type: bool
    default: false
notes:
  - Requires LVM2 utilities installed on the target system.
  - Device path must exist when creating a PV.
"""

EXAMPLES = r"""
- name: Creating physical volume on /dev/sdb
  community.general.lvm_pv:
    device: /dev/sdb

- name: Creating and resizing (if needed) physical volume
  community.general.lvm_pv:
    device: /dev/sdb
    resize: true

- name: Removing physical volume that is not part of any volume group
  community.general.lvm_pv:
    device: /dev/sdb
    state: absent

- name: Force removing physical volume that is already part of a volume group
  community.general.lvm_pv:
    device: /dev/sdb
    force: true
    state: absent
"""

RETURN = r"""
"""


import os
from ansible.module_utils.basic import AnsibleModule


def get_pv_status(module, device):
    """Check if the device is already a PV."""
    cmd = ['pvs', '--noheadings', '--readonly', device]
    return module.run_command(cmd)[0] == 0


def get_pv_size(module, device):
    """Get current PV size in bytes."""
    cmd = ['pvs', '--noheadings', '--nosuffix', '--units', 'b', '-o', 'pv_size', device]
    rc, out, err = module.run_command(cmd, check_rc=True)
    return int(out.strip())


def rescan_device(module, device):
    """Perform storage rescan for the device."""
    # Extract the base device name (e.g., /dev/sdb -> sdb)
    base_device = os.path.basename(device)
    rescan_path = "/sys/block/{0}/device/rescan".format(base_device)

    if os.path.exists(rescan_path):
        try:
            with open(rescan_path, 'w') as f:
                f.write('1')
            return True
        except IOError as e:
            module.warn("Failed to rescan device {0}: {1}".format(device, str(e)))
            return False
    else:
        module.warn("Rescan path not found for device {0}".format(device))
        return False


def main():
    module = AnsibleModule(
        argument_spec=dict(
            device=dict(type='path', required=True),
            state=dict(type='str', default='present', choices=['present', 'absent']),
            force=dict(type='bool', default=False),
            resize=dict(type='bool', default=False),
        ),
        supports_check_mode=True,
    )

    device = module.params['device']
    state = module.params['state']
    force = module.params['force']
    resize = module.params['resize']
    changed = False
    actions = []

    # Validate device existence for present state
    if state == 'present' and not os.path.exists(device):
        module.fail_json(msg="Device %s not found" % device)

    is_pv = get_pv_status(module, device)

    if state == 'present':
        # Create PV if needed
        if not is_pv:
            if module.check_mode:
                changed = True
                actions.append('would be created')
            else:
                cmd = ['pvcreate']
                if force:
                    cmd.append('-f')
                cmd.append(device)
                rc, out, err = module.run_command(cmd, check_rc=True)
                changed = True
                actions.append('created')
            is_pv = True

        # Handle resizing
        elif resize and is_pv:
            if module.check_mode:
                # In check mode, assume resize would change
                changed = True
                actions.append('would be resized')
            else:
                # Perform device rescan if each time
                if rescan_device(module, device):
                    actions.append('rescanned')
                original_size = get_pv_size(module, device)
                rc, out, err = module.run_command(['pvresize', device], check_rc=True)
                new_size = get_pv_size(module, device)
                if new_size != original_size:
                    changed = True
                    actions.append('resized')

    elif state == 'absent':
        if is_pv:
            if module.check_mode:
                changed = True
                actions.append('would be removed')
            else:
                cmd = ['pvremove', '-y']
                if force:
                    cmd.append('-ff')
                changed = True
                cmd.append(device)
                rc, out, err = module.run_command(cmd, check_rc=True)
                actions.append('removed')

    # Generate final message
    if actions:
        msg = "PV %s: %s" % (device, ', '.join(actions))
    else:
        msg = "No changes needed for PV %s" % device
    module.exit_json(changed=changed, msg=msg)


if __name__ == '__main__':
    main()