Ecrire un module Ansible personnalise

Anatomie d'un module Ansible

Un module Ansible est un script Python qui recoit des arguments en JSON et retourne un resultat structure.

Module simple en Python

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

from ansible.module_utils.basic import AnsibleModule

DOCUMENTATION = r'''
---
module: mon_service
short_description: Gere un service personnalise
description:
  - Ce module permet de demarrer, arreter ou redemarrer un service personnalise.
options:
  name:
    description: Nom du service
    required: true
    type: str
  state:
    description: Etat souhaite
    choices: [started, stopped, restarted]
    default: started
    type: str
author:
  - Equipe DevOps
'''

EXAMPLES = r'''
- name: Demarrer mon service
  mon_entreprise.infra.mon_service:
    name: mon-app
    state: started

- name: Redemarrer mon service
  mon_entreprise.infra.mon_service:
    name: mon-app
    state: restarted
'''

RETURN = r'''
status:
  description: Statut actuel du service
  returned: always
  type: str
  sample: running
pid:
  description: PID du processus
  returned: when started
  type: int
  sample: 12345
'''


def run_module():
    module_args = dict(
        name=dict(type='str', required=True),
        state=dict(
            type='str',
            default='started',
            choices=['started', 'stopped', 'restarted']
        ),
    )

    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True,
    )

    name = module.params['name']
    state = module.params['state']
    changed = False
    result = {'name': name, 'state': state}

    # Mode check : simuler sans appliquer
    if module.check_mode:
        module.exit_json(changed=True, **result)

    # Logique du module
    rc, stdout, stderr = module.run_command(
        ['systemctl', 'is-active', name]
    )
    current_state = stdout.strip()

    if state == 'started' and current_state != 'active':
        rc, stdout, stderr = module.run_command(
            ['systemctl', 'start', name]
        )
        if rc != 0:
            module.fail_json(msg=f'Impossible de demarrer {name}: {stderr}')
        changed = True
    elif state == 'stopped' and current_state == 'active':
        rc, stdout, stderr = module.run_command(
            ['systemctl', 'stop', name]
        )
        if rc != 0:
            module.fail_json(msg=f'Impossible d''arreter {name}: {stderr}')
        changed = True
    elif state == 'restarted':
        rc, stdout, stderr = module.run_command(
            ['systemctl', 'restart', name]
        )
        if rc != 0:
            module.fail_json(msg=f'Impossible de redemarrer {name}: {stderr}')
        changed = True

    result['status'] = 'running' if state != 'stopped' else 'stopped'
    module.exit_json(changed=changed, **result)


def main():
    run_module()


if __name__ == '__main__':
    main()

Tester un module

# Test local avec ansible-doc
ansible-doc -M ./plugins/modules mon_service

# Test avec un playbook
- hosts: localhost
  tasks:
    - name: Tester mon module
      mon_entreprise.infra.mon_service:
        name: nginx
        state: started
      register: result

    - debug:
        var: result

Plugins de filtre personnalises

# plugins/filter/mes_filtres.py
class FilterModule:
    def filters(self):
        return {
            'to_nginx_upstream': self.to_nginx_upstream,
        }

    def to_nginx_upstream(self, servers, port=80):
        lines = []
        for server in servers:
            lines.append(f'    server {server}:{port};')
        return '\n'.join(lines)

# Utilisation dans un template :
# upstream backend {
# {{ backend_servers | to_nginx_upstream(8080) }}
# }