ceph_config.py

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

# Copyright Red Hat
# SPDX-License-Identifier: Apache-2.0
# Author: Guillaume Abrioux <gabrioux@redhat.com>

from __future__ import absolute_import, division, print_function
__metaclass__ = type

ANSIBLE_METADATA = {
    'metadata_version': '1.1',
    'status': ['preview'],
    'supported_by': 'community'
}

DOCUMENTATION = '''
---
module: ceph_config
short_description: set ceph config
version_added: "1.1.0"
description:
    - Set Ceph config options.
options:
    fsid:
        description:
            - the fsid of the Ceph cluster to interact with.
        type: str
        required: false
    image:
        description:
            - The Ceph container image to use.
        type: str
        required: false
    action:
        description:
            - whether to get, set, or remove the parameter specified in 'option'
        type: str
        choices: ['get', 'set','remove']
        default: 'set'
        required: false
    who:
        description:
            - which daemon the configuration should be set to
        type: str
        required: true
    option:
        description:
            - name of the parameter to be set
        type: str
        required: true
    value:
        description:
            - value of the parameter
        type: str
        required: false

author:
    - guillaume abrioux (@guits)
'''

EXAMPLES = '''
- name: set osd_memory_target for osd.0
  ceph_config:
    action: set
    who: osd.0
    option: osd_memory_target
    value: 5368709120

- name: set osd_memory_target for host ceph-osd-02
  ceph_config:
    action: set
    who: osd/host:ceph-osd-02
    option: osd_memory_target
    value: 5368709120

- name: get osd_pool_default_size value
  ceph_config:
    action: get
    who: global
    option: osd_pool_default_size
    

- name: remove osd_memory_target override
  ceph_config:
    action: remove
    who: osd
    option: osd_memory_target

'''

RETURN = '''#  '''

from typing import Any, Dict, List, Tuple, Union
from ansible.module_utils.basic import AnsibleModule  # type: ignore
try:
    from ansible_collections.ceph.automation.plugins.module_utils.ceph_common import exit_module, build_base_cmd_shell, fatal  # type: ignore
except ImportError:
    from module_utils.ceph_common import exit_module, build_base_cmd_shell, fatal  # type: ignore

import datetime
import json


def set_option(module: "AnsibleModule",
               who: str,
               option: str,
               value: str) -> Tuple[int, List[str], str, str]:
    cmd = build_base_cmd_shell(module)
    cmd.extend(['ceph', 'config', 'set', who, option, value])

    rc, out, err = module.run_command(cmd)

    return rc, cmd, out.strip(), err
def remove_option(module: "AnsibleModule",
                  who: str,
                  option: str) -> Tuple[int, List[str], str, str]:
    cmd = build_base_cmd_shell(module)
    cmd.extend(['ceph', 'config', 'rm', who, option])
    rc, out, err = module.run_command(cmd)
    return rc, cmd, out.strip(), err

  

def get_config_dump(module: "AnsibleModule") -> Tuple[int, List[str], str, str]:
    cmd = build_base_cmd_shell(module)
    cmd.extend(['ceph', 'config', 'dump', '--format', 'json'])
    rc, out, err = module.run_command(cmd)
    if rc:
        fatal(message=f"Can't get current configuration via `ceph config dump`.Error:\n{err}", module=module)
    out = out.strip()
    return rc, cmd, out, err


def get_current_value(who: str, option: str, config_dump: List[Dict[str, Any]]) -> Union[str, None]:
    for config in config_dump:
        if config['section'] == who and config['name'] == option:
            return config['value']
    return None


def main() -> None:
    module = AnsibleModule(
        argument_spec=dict(
            who=dict(type='str', required=True),
            action=dict(type='str', required=False, choices=['get', 'set','remove'], default='set'),
            option=dict(type='str', required=True),
            value=dict(type='str', required=False),
            fsid=dict(type='str', required=False),
            image=dict(type='str', required=False)
        ),
        supports_check_mode=True,
        required_if=[['action', 'set', ['value']]]
    )

    # Gather module parameters in variables
    who = module.params.get('who')
    option = module.params.get('option')
    value = module.params.get('value')
    action = module.params.get('action')

    startd = datetime.datetime.now()
    changed = False

    rc, cmd, out, err = get_config_dump(module)
    config_dump = json.loads(out)
    current_value = get_current_value(who, option, config_dump)

    if action == 'set':
        if str(value).lower() == str(current_value).lower():
            out = 'who={} option={} value={} already set. Skipping.'.format(who, option, value)
        else:
            if module.check_mode:
                out = 'who={} option={} would be set to {}'.format(who, option, value)
                changed = True
            else:
                rc, cmd, out, err = set_option(module, who, option, value)
                if rc != 0:
                    module.fail_json(msg=err, cmd=cmd, rc=rc)
                changed = True

    elif action == 'remove':
        if current_value is None:
            out = 'who={} option={} already absent. Skipping.'.format(who, option)
        else:
            if module.check_mode:
                out = 'who={} option={} would be removed'.format(who, option)
                changed = True
            else:
                rc, cmd, out, err = remove_option(module, who, option)
                if rc != 0:
                    module.fail_json(msg=err, cmd=cmd, rc=rc)
                changed = True
    else:
        if current_value is None:
            out = ''
            err = 'No value found for who={} option={}'.format(who, option)
        else:
            out = current_value

    exit_module(module=module, out=out, rc=rc,
                cmd=cmd, err=err, startd=startd,
                changed=changed)


if __name__ == '__main__':
    main()