Eigene Module

Eigene Module erweitern Ansible um spezifische Funktionalitäten, die durch Built-in-Module nicht abgedeckt werden. Sie ermöglichen die Integration proprietärer Systeme, Legacy-Anwendungen oder spezieller APIs in die Ansible-Automatisierung.

37.1.2 Unterschiede zu Built-in- und Community-Modulen

Built-in-Module: Teil des Ansible-Core, universell verfügbar, umfassend getestet.

Community-Module: Über Collections verfügbar, von der Community entwickelt und gepflegt.

Eigene Module: Projektspezifisch entwickelt, vollständige Kontrolle über Funktionalität und Wartung.

37.1.3 Typische Einsatzszenarien

# Beispiel: Interne Monitoring-API
def call_monitoring_api(endpoint, data):
    # Custom API-Logik
    pass

Proprietäre Schnittstellen: Integration unternehmensinterner Tools und Plattformen.

Komplexe Geschäftslogik: Implementierung spezifischer Arbeitsabläufe, die mehrere Systemaufrufe kombinieren.

37.2 Aufbau eines Moduls

37.2.1 Verzeichnisstruktur

37.2.2 Grundgerüst in Python

#!/usr/bin/python

from ansible.module_utils.basic import AnsibleModule

def main():
    # Modul-Definition
    module = AnsibleModule(
        argument_spec={}
    )
    
    # Modul-Logik
    result = dict(changed=False)
    
    # Erfolgreiche Rückgabe
    module.exit_json(**result)

if __name__ == '__main__':
    main()

37.2.3 Argument-Spezifikation

argument_spec = dict(
    name=dict(type='str', required=True),
    state=dict(type='str', default='present', choices=['present', 'absent']),
    port=dict(type='int', default=80),
    enabled=dict(type='bool', default=True),
    tags=dict(type='list', elements='str', default=[])
)

Parameter-Typen: - str: Zeichenkette - int: Ganzzahl - bool: Boolean (True/False) - list: Liste von Elementen - dict: Dictionary-Objekt - path: Dateipfad

Parameter-Optionen: - required: Parameter ist obligatorisch - default: Standardwert - choices: Erlaubte Werte - elements: Typ der Listen-Elemente

37.2.4 Rückgabewerte

result = dict(
    changed=True,
    msg="Operation completed successfully",
    data={"key": "value"}
)
module.exit_json(**result)

module.fail_json(
    msg="Operation failed: Invalid parameter",
    error_code=400
)

Standard-Rückgabefelder: - changed: Wurde das System verändert? - msg: Statusmeldung - failed: Ist ein Fehler aufgetreten? - Beliebige zusätzliche Datenfelder

37.3 Minimalbeispiel

37.3.1 Hello World-Modul

#!/usr/bin/python

from ansible.module_utils.basic import AnsibleModule

def main():
    # Argument-Definition
    module_args = dict(
        name=dict(type='str', required=True),
        greeting=dict(type='str', default='Hello')
    )
    
    # Modul erstellen
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )
    
    # Parameter auslesen
    name = module.params['name']
    greeting = module.params['greeting']
    
    # Geschäftslogik
    message = f"{greeting}, {name}!"
    
    # Ergebnis zusammenstellen
    result = dict(
        changed=False,
        message=message,
        original_name=name,
        greeting_used=greeting
    )
    
    # Check-Mode unterstützen
    if module.check_mode:
        module.exit_json(**result)
    
    # Erfolgreiche Rückgabe
    module.exit_json(**result)

if __name__ == '__main__':
    main()

37.3.2 Playbook für Hello World-Modul

---
- hosts: localhost
  connection: local
  gather_facts: no
  
  tasks:
    - name: Standard-Begrüßung
      custom_hello:
        name: "Ansible User"
      register: hello_result
      
    - name: Ergebnis anzeigen
      debug:
        var: hello_result
        
    - name: Benutzerdefinierte Begrüßung
      custom_hello:
        name: "DevOps Team"
        greeting: "Willkommen"
      register: custom_hello
      
    - name: Benutzerdefiniertes Ergebnis
      debug:
        msg: "{{ custom_hello.message }}"

ansible-playbook test-hello.yml

37.4 Erweiterung

37.4.1 Parameter verarbeiten

def validate_parameters(module):
    """Validiert Eingabeparameter."""
    port = module.params['port']
    
    if port < 1 or port > 65535:
        module.fail_json(
            msg=f"Port {port} ist außerhalb des gültigen Bereichs (1-65535)"
        )
    
    name = module.params['name']
    if not name.isalnum():
        module.fail_json(
            msg="Name darf nur alphanumerische Zeichen enthalten"
        )

def main():
    module_args = dict(
        name=dict(type='str', required=True),
        port=dict(type='int', default=8080),
        config=dict(type='dict', default={})
    )
    
    module = AnsibleModule(argument_spec=module_args)
    
    # Parameter validieren
    validate_parameters(module)
    
    # Weitere Verarbeitung...

37.4.2 Fehlerbedingungen implementieren

import traceback

def execute_operation(module):
    """Führt die Hauptoperation aus."""
    try:
        # Risikoreiche Operation
        result = perform_system_call()
        return result
        
    except FileNotFoundError as e:
        module.fail_json(
            msg=f"Datei nicht gefunden: {str(e)}",
            error_type="file_not_found",
            path=e.filename
        )
        
    except PermissionError as e:
        module.fail_json(
            msg=f"Berechtigung verweigert: {str(e)}",
            error_type="permission_denied"
        )
        
    except Exception as e:
        module.fail_json(
            msg=f"Unerwarteter Fehler: {str(e)}",
            error_type="unexpected_error",
            exception=traceback.format_exc()
        )

37.4.3 Idempotenz berücksichtigen

def check_current_state(module):
    """Prüft aktuellen Systemzustand."""
    target_file = module.params['path']
    desired_content = module.params['content']
    
    try:
        with open(target_file, 'r') as f:
            current_content = f.read()
        
        return {
            'exists': True,
            'content_matches': current_content == desired_content,
            'current_content': current_content
        }
    except FileNotFoundError:
        return {
            'exists': False,
            'content_matches': False,
            'current_content': None
        }

def main():
    module = AnsibleModule(argument_spec=module_args)
    
    # Aktuellen Zustand prüfen
    current_state = check_current_state(module)
    
    # Nur ändern wenn nötig
    if current_state['content_matches']:
        module.exit_json(
            changed=False,
            msg="Datei bereits im gewünschten Zustand"
        )
    
    # Änderung durchführen
    perform_change(module)
    
    module.exit_json(
        changed=True,
        msg="Datei erfolgreich aktualisiert"
    )

37.5 Übungen

37.5.1 Übung 1: Minimalmodul erstellen (echo)

#!/usr/bin/python

from ansible.module_utils.basic import AnsibleModule

def main():
    # Parameter definieren
    module_args = dict(
        text=dict(type='str', required=True),
        uppercase=dict(type='bool', default=False)
    )
    
    # Modul erstellen
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )
    
    # Parameter auslesen
    text = module.params['text']
    uppercase = module.params['uppercase']
    
    # Text verarbeiten
    output_text = text.upper() if uppercase else text
    
    # Ergebnis
    result = dict(
        changed=False,
        echo=output_text,
        original=text,
        was_uppercase=uppercase
    )
    
    module.exit_json(**result)

if __name__ == '__main__':
    main()

---
- hosts: localhost
  connection: local
  gather_facts: no
  
  tasks:
    - name: Text normal ausgeben
      simple_echo:
        text: "Hallo Ansible!"
      register: echo_normal
      
    - name: Normales Echo anzeigen
      debug:
        msg: "Echo: {{ echo_normal.echo }}"
        
    - name: Text in Großbuchstaben
      simple_echo:
        text: "Hallo Ansible!"
        uppercase: true
      register: echo_upper
      
    - name: Großbuchstaben-Echo anzeigen
      debug:
        msg: "Echo: {{ echo_upper.echo }}"

37.5.2 Übung 2: Modul mit Parametern (file_checker)

#!/usr/bin/python

import os
import stat
from ansible.module_utils.basic import AnsibleModule

def check_file_properties(filepath):
    """Prüft Eigenschaften einer Datei."""
    if not os.path.exists(filepath):
        return {
            'exists': False,
            'error': 'Datei existiert nicht'
        }
    
    try:
        file_stat = os.stat(filepath)
        
        return {
            'exists': True,
            'size': file_stat.st_size,
            'mode': oct(stat.S_IMODE(file_stat.st_mode)),
            'is_file': os.path.isfile(filepath),
            'is_directory': os.path.isdir(filepath),
            'readable': os.access(filepath, os.R_OK),
            'writable': os.access(filepath, os.W_OK),
            'executable': os.access(filepath, os.X_OK)
        }
        
    except Exception as e:
        return {
            'exists': True,
            'error': f'Fehler beim Lesen der Eigenschaften: {str(e)}'
        }

def main():
    module_args = dict(
        path=dict(type='str', required=True),
        check_content=dict(type='bool', default=False),
        max_size=dict(type='int', default=None)
    )
    
    module = AnsibleModule(
        argument_spec=module_args,
        supports_check_mode=True
    )
    
    filepath = module.params['path']
    check_content = module.params['check_content']
    max_size = module.params['max_size']
    
    # Dateieigenschaften prüfen
    file_props = check_file_properties(filepath)
    
    if 'error' in file_props:
        module.fail_json(msg=file_props['error'])
    
    # Größe prüfen falls gewünscht
    if max_size and file_props['exists'] and file_props['size'] > max_size:
        module.fail_json(
            msg=f"Datei zu groß: {file_props['size']} Bytes (Maximum: {max_size})"
        )
    
    # Inhalt lesen falls gewünscht
    content_info = {}
    if check_content and file_props['exists'] and file_props['is_file']:
        try:
            with open(filepath, 'r') as f:
                content = f.read()
                content_info = {
                    'line_count': len(content.splitlines()),
                    'char_count': len(content),
                    'first_line': content.splitlines()[0] if content.splitlines() else ''
                }
        except Exception as e:
            content_info = {'error': f'Fehler beim Lesen: {str(e)}'}
    
    # Ergebnis zusammenstellen
    result = dict(
        changed=False,
        path=filepath,
        properties=file_props,
        content_info=content_info if check_content else None
    )
    
    module.exit_json(**result)

if __name__ == '__main__':
    main()

---
- hosts: localhost
  connection: local
  gather_facts: no
  
  tasks:
    - name: Test-Datei erstellen
      copy:
        content: |
          Zeile 1
          Zeile 2
          Zeile 3
        dest: /tmp/test-file.txt
        
    - name: Datei-Eigenschaften prüfen
      file_checker:
        path: /tmp/test-file.txt
        check_content: true
        max_size: 1000
      register: file_check
      
    - name: Ergebnis anzeigen
      debug:
        var: file_check
        
    - name: Nicht-existierende Datei prüfen
      file_checker:
        path: /tmp/does-not-exist.txt
      register: missing_file
      ignore_errors: yes
      
    - name: Fehler-Ergebnis anzeigen
      debug:
        var: missing_file

37.5.3 Übung 3: Komplexes Modul mit Fehlerbehandlung (service_manager)

Aufgabe: Erstelle ein Modul, das einen einfachen Service verwaltet (Start/Stop einer Anwendung).

#!/usr/bin/python

import os
import signal
import time
import subprocess
from ansible.module_utils.basic import AnsibleModule

class ServiceManager:
    def __init__(self, module):
        self.module = module
        self.name = module.params['name']
        self.command = module.params['command']
        self.pidfile = module.params.get('pidfile', f'/tmp/{self.name}.pid')
        self.working_dir = module.params.get('working_dir', '/tmp')
        
    def is_running(self):
        """Prüft ob Service läuft."""
        if not os.path.exists(self.pidfile):
            return False
            
        try:
            with open(self.pidfile, 'r') as f:
                pid = int(f.read().strip())
            
            # Prüfen ob Prozess existiert
            os.kill(pid, 0)
            return True
            
        except (ValueError, ProcessLookupError, OSError):
            # PID-File löschen wenn Prozess nicht existiert
            try:
                os.remove(self.pidfile)
            except OSError:
                pass
            return False
    
    def start_service(self):
        """Startet den Service."""
        if self.is_running():
            return {
                'changed': False,
                'msg': f'Service {self.name} läuft bereits'
            }
        
        try:
            # Service im Hintergrund starten
            process = subprocess.Popen(
                self.command,
                shell=True,
                cwd=self.working_dir,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL
            )
            
            # PID speichern
            with open(self.pidfile, 'w') as f:
                f.write(str(process.pid))
            
            # Kurz warten und prüfen ob Service noch läuft
            time.sleep(1)
            if not self.is_running():
                return {
                    'changed': False,
                    'failed': True,
                    'msg': f'Service {self.name} konnte nicht gestartet werden'
                }
            
            return {
                'changed': True,
                'msg': f'Service {self.name} erfolgreich gestartet',
                'pid': process.pid
            }
            
        except Exception as e:
            return {
                'changed': False,
                'failed': True,
                'msg': f'Fehler beim Starten des Services: {str(e)}'
            }
    
    def stop_service(self):
        """Stoppt den Service."""
        if not self.is_running():
            return {
                'changed': False,
                'msg': f'Service {self.name} läuft nicht'
            }
        
        try:
            with open(self.pidfile, 'r') as f:
                pid = int(f.read().strip())
            
            # Graceful shutdown versuchen
            os.kill(pid, signal.SIGTERM)
            
            # Warten bis Prozess beendet ist
            for _ in range(10):  # Max 10 Sekunden warten
                try:
                    os.kill(pid, 0)
                    time.sleep(1)
                except ProcessLookupError:
                    break
            else:
                # Force kill wenn graceful shutdown fehlschlägt
                try:
                    os.kill(pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass
            
            # PID-File entfernen
            try:
                os.remove(self.pidfile)
            except OSError:
                pass
            
            return {
                'changed': True,
                'msg': f'Service {self.name} erfolgreich gestoppt'
            }
            
        except Exception as e:
            return {
                'changed': False,
                'failed': True,
                'msg': f'Fehler beim Stoppen des Services: {str(e)}'
            }
    
    def get_status(self):
        """Gibt Service-Status zurück."""
        running = self.is_running()
        pid = None
        
        if running and os.path.exists(self.pidfile):
            try:
                with open(self.pidfile, 'r') as f:
                    pid = int(f.read().strip())
            except (ValueError, OSError):
                pass
        
        return {
            'changed': False,
            'running': running,
            'pid': pid,
            'pidfile': self.pidfile,
            'name': self.name
        }

def main():
    module_args = dict(
        name=dict(type='str', required=True),
        command=dict(type='str', required=False),
        state=dict(type='str', default='started', choices=['started', 'stopped', 'status']),
        pidfile=dict(type='str', required=False),
        working_dir=dict(type='str', default='/tmp')
    )
    
    module = AnsibleModule(
        argument_spec=module_args,
        required_if=[
            ('state', 'started', ['command']),
        ],
        supports_check_mode=True
    )
    
    service_mgr = ServiceManager(module)
    state = module.params['state']
    
    # Check Mode
    if module.check_mode:
        if state == 'status':
            result = service_mgr.get_status()
        else:
            result = {
                'changed': True,
                'msg': f'Would {state} service {service_mgr.name}'
            }
        module.exit_json(**result)
    
    # Normale Ausführung
    if state == 'started':
        result = service_mgr.start_service()
    elif state == 'stopped':
        result = service_mgr.stop_service()
    elif state == 'status':
        result = service_mgr.get_status()
    else:
        module.fail_json(msg=f'Unbekannter Zustand: {state}')
    
    # Fehlerbehandlung
    if result.get('failed', False):
        module.fail_json(**result)
    
    module.exit_json(**result)

if __name__ == '__main__':
    main()

---
- hosts: localhost
  connection: local
  gather_facts: no
  
  tasks:
    - name: Test-Script erstellen
      copy:
        content: |
          #!/bin/bash
          echo "Test-Service gestartet: $(date)" >> /tmp/test-service.log
          while true; do
            sleep 5
            echo "Service läuft: $(date)" >> /tmp/test-service.log
          done
        dest: /tmp/test-service.sh
        mode: '0755'
        
    - name: Service-Status prüfen
      service_manager:
        name: test-service
        state: status
      register: initial_status
      
    - name: Initialer Status
      debug:
        var: initial_status
        
    - name: Service starten
      service_manager:
        name: test-service
        command: "/tmp/test-service.sh"
        state: started
        pidfile: /tmp/test-service.pid
      register: start_result
      
    - name: Start-Ergebnis
      debug:
        var: start_result
        
    - name: Service-Status nach Start prüfen
      service_manager:
        name: test-service
        state: status
      register: running_status
      
    - name: Laufender Status
      debug:
        var: running_status
        
    - name: Service nochmals starten (Idempotenz-Test)
      service_manager:
        name: test-service
        command: "/tmp/test-service.sh"
        state: started
      register: idempotent_start
      
    - name: Idempotenz-Ergebnis
      debug:
        var: idempotent_start
        
    - name: Log-Datei prüfen
      command: tail -5 /tmp/test-service.log
      register: log_content
      ignore_errors: yes
      
    - name: Log-Inhalt anzeigen
      debug:
        var: log_content.stdout_lines
      when: log_content.rc == 0
        
    - name: Service stoppen
      service_manager:
        name: test-service
        state: stopped
      register: stop_result
      
    - name: Stop-Ergebnis
      debug:
        var: stop_result
        
    - name: Service-Status nach Stop prüfen
      service_manager:
        name: test-service
        state: status
      register: final_status
      
    - name: Finaler Status
      debug:
        var: final_status

37.6 Best Practices

37.6.1 Namenskonventionen

Modul-Namen: - Verwende aussagekräftige Namen: user_manager statt um - Trenne Wörter mit Unterstrichen: api_client statt apiclient - Vermeide Konflikte mit Built-in-Modulen

Parameter-Namen: - Verwende konsistente Namenskonventionen - Standard-Parameter wie state, name, path wenn möglich - Dokumentiere komplexe Parameter ausführlich

37.6.2 Logging und Debugging

from ansible.module_utils.basic import AnsibleModule

def main():
    module = AnsibleModule(argument_spec=module_args)
    
    # Debug-Modus erkennen
    if module._verbosity >= 3:
        module.log(f"Debug: Verarbeite Parameter {module.params}")
    
    # Wichtige Zwischenergebnisse loggen
    module.log(f"Info: Operation started for {module.params['name']}")

import traceback

try:
    risky_operation()
except Exception as e:
    module.fail_json(
        msg=f"Operation fehlgeschlagen: {str(e)}",
        exception=traceback.format_exc(),
        module_params=module.params
    )

37.6.3 Verteilung über Collections

namespace: company
name: infrastructure
version: 1.0.0
description: Custom infrastructure modules
authors:
  - IT Department <it@company.com>
dependencies: {}

# Collection bauen
ansible-galaxy collection build

# Collection installieren
ansible-galaxy collection install company-infrastructure-1.0.0.tar.gz

- name: Benutzer verwalten
  company.infrastructure.user_manager:
    name: testuser
    state: present

37.6.4 Dokumentation

#!/usr/bin/python

DOCUMENTATION = r'''
---
module: service_manager
short_description: Verwaltet einfache Services
description:
    - Startet, stoppt und überwacht einfache Services
    - Verwendet PID-Files für Zustandsverfolgung
options:
    name:
        description: Name des Services
        required: true
        type: str
    command:
        description: Befehl zum Starten des Services
        required: false
        type: str
    state:
        description: Gewünschter Service-Zustand
        choices: [started, stopped, status]
        default: started
        type: str
'''

EXAMPLES = r'''
- name: Service starten
  service_manager:
    name: myapp
    command: "/usr/bin/myapp --daemon"
    state: started

- name: Service stoppen
  service_manager:
    name: myapp
    state: stopped
'''

RETURN = r'''
changed:
    description: Ob Änderungen vorgenommen wurden
    type: bool
    returned: always
running:
    description: Ob der Service läuft
    type: bool
    returned: when state=status
pid:
    description: Prozess-ID des Services
    type: int
    returned: when running
'''

37.7 Ausblick

37.7.1 API-Integration

import requests
from ansible.module_utils.basic import AnsibleModule

def api_request(module, endpoint, method='GET', data=None):
    """Führt API-Request aus."""
    base_url = module.params['api_url']
    headers = {
        'Authorization': f"Bearer {module.params['api_token']}",
        'Content-Type': 'application/json'
    }
    
    try:
        response = requests.request(
            method=method,
            url=f"{base_url}/{endpoint}",
            headers=headers,
            json=data,
            timeout=30
        )
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.RequestException as e:
        module.fail_json(
            msg=f"API-Request fehlgeschlagen: {str(e)}",
            status_code=getattr(e.response, 'status_code', None)
        )

37.7.2 Komplexe Systeme

import psycopg2
from ansible.module_utils.basic import AnsibleModule

def connect_database(module):
    """Stellt Datenbankverbindung her."""
    try:
        conn = psycopg2.connect(
            host=module.params['db_host'],
            database=module.params['db_name'],
            user=module.params['db_user'],
            password=module.params['db_password']
        )
        return conn
    except psycopg2.Error as e:
        module.fail_json(msg=f"Datenbankverbindung fehlgeschlagen: {str(e)}")

def orchestrate_deployment(module):
    """Orchestriert komplexe Deployment-Schritte."""
    steps = [
        ('database', update_database_schema),
        ('cache', clear_application_cache),
        ('services', restart_application_services),
        ('monitoring', update_monitoring_config)
    ]
    
    results = {}
    for step_name, step_function in steps:
        try:
            result = step_function(module)
            results[step_name] = result
        except Exception as e:
            # Rollback vorheriger Schritte
            rollback_steps(module, results)
            module.fail_json(
                msg=f"Deployment fehlgeschlagen bei Schritt {step_name}: {str(e)}",
                completed_steps=list(results.keys())
            )
    
    return results

Eigene Module erweitern Ansible um projektspezifische Funktionalitäten und ermöglichen die Integration beliebiger Systeme in die Automatisierung. Sie folgen denselben Prinzipien wie Built-in-Module und können über Collections verteilt und wiederverwendet werden.

37 Eigene Module

37.1 Einführung

37.1.1 Warum eigene Module?