43 Eigene Plugins entwickeln

# plugins/filter/string_utils.py
from ansible.errors import AnsibleError
import re

def to_slug(value, separator='-'):
    """
    Konvertiert String zu URL-sicherem Slug
    'My Service #2!' -> 'my-service-2'
    """
    if not isinstance(value, str):
        raise AnsibleError(f"to_slug erwartet String, erhielt {type(value)}")
    
    # Zu Kleinbuchstaben, Sonderzeichen entfernen
    slug = re.sub(r'[^\w\s-]', '', value.lower())
    # Leerzeichen durch Separator ersetzen  
    slug = re.sub(r'[\s_-]+', separator, slug)
    return slug.strip(separator)

class FilterModule(object):
    def filters(self):
        return {'to_slug': to_slug}

mkdir -p plugins/filter
# Plugin-Datei erstellen (siehe oben)

---
- name: Test eigener String-Filter
  hosts: localhost
  gather_facts: false
  vars:
    service_names:
      - "My Awesome Service!"
      - "Database Cache #2"
      - "Queue & Worker System"

  tasks:
    - name: URL-sichere Namen generieren
      debug:
        msg: "{{ item }} -> {{ item | to_slug }}"
      loop: "{{ service_names }}"

    - name: Mit anderem Separator
      debug:
        msg: "{{ item }} -> {{ item | to_slug('_') }}"
      loop: "{{ service_names }}"

    - name: Praktische Anwendung
      debug:
        msg: "Verzeichnis: /opt/services/{{ item | to_slug }}"
      loop: "{{ service_names }}"

ansible-playbook test-string-filter.yml

# plugins/lookup/config_file.py
from ansible.plugins.lookup import LookupBase
from ansible.errors import AnsibleError
import os
import json

class LookupModule(LookupBase):
    """
    Lädt JSON-Konfigurationsdateien aus mehreren Suchpfaden
    {{ lookup('config_file', 'database.json') }}
    """
    
    def run(self, terms, variables=None, **kwargs):
        search_paths = kwargs.get('search_paths', [
            './configs',
            '/etc/ansible/configs',
            os.path.expanduser('~/.ansible/configs')
        ])
        
        results = []
        
        for term in terms:
            found = False
            
            for search_path in search_paths:
                full_path = os.path.join(search_path, term)
                
                if os.path.exists(full_path):
                    try:
                        with open(full_path, 'r') as f:
                            data = json.load(f)
                        results.append(data)
                        found = True
                        break
                    except (IOError, json.JSONDecodeError) as e:
                        raise AnsibleError(f"Fehler beim Lesen von {full_path}: {e}")
            
            if not found:
                raise AnsibleError(f"Konfigurationsdatei '{term}' nicht gefunden in: {search_paths}")
        
        return results

mkdir -p configs
cat > configs/database.json << EOF
{
  "host": "localhost",
  "port": 5432,
  "database": "myapp",
  "username": "dbuser"
}
EOF

cat > configs/redis.json << EOF
{
  "host": "redis.company.com", 
  "port": 6379,
  "db": 0
}
EOF

---
- name: Test Config-Lookup Plugin
  hosts: localhost
  gather_facts: false
  
  tasks:
    - name: Datenbank-Konfiguration laden
      debug:
        var: db_config
      vars:
        db_config: "{{ lookup('config_file', 'database.json') }}"

    - name: Redis-Konfiguration laden
      debug:
        msg: "Redis läuft auf {{ redis_config.host }}:{{ redis_config.port }}"
      vars:
        redis_config: "{{ lookup('config_file', 'redis.json') }}"

    - name: Mit benutzerdefinierten Suchpfaden
      debug:
        var: custom_config
      vars:
        custom_config: "{{ lookup('config_file', 'database.json', search_paths=['./configs', '/tmp']) }}"

ansible-playbook test-config-lookup.yml

# plugins/callback/json_logger.py
from ansible.plugins.callback import CallbackBase
import json
import os
from datetime import datetime

class CallbackModule(CallbackBase):
    """
    Schreibt Ansible-Events als JSON in Logdatei
    """
    
    CALLBACK_VERSION = 2.0
    CALLBACK_TYPE = 'stdout'
    CALLBACK_NAME = 'json_logger'
    
    def __init__(self):
        super(CallbackModule, self).__init__()
        self.log_file = os.getenv('ANSIBLE_JSON_LOG', '/tmp/ansible_events.log')
        self.start_time = None

    def v2_playbook_on_start(self, playbook):
        self.start_time = datetime.utcnow()
        self._log_event('playbook_start', {
            'playbook': os.path.basename(playbook._file_name),
            'user': os.getenv('USER', 'unknown'),
            'timestamp': self.start_time.isoformat()
        })

    def v2_runner_on_ok(self, result):
        self._log_event('task_success', {
            'host': result._host.name,
            'task': result._task.get_name(),
            'changed': result._result.get('changed', False),
            'timestamp': datetime.utcnow().isoformat()
        })

    def v2_runner_on_failed(self, result, ignore_errors=False):
        self._log_event('task_failed', {
            'host': result._host.name,
            'task': result._task.get_name(),
            'error': result._result.get('msg', 'Unknown error'),
            'ignore_errors': ignore_errors,
            'timestamp': datetime.utcnow().isoformat()
        })

    def v2_playbook_on_stats(self, stats):
        duration = (datetime.utcnow() - self.start_time).total_seconds()
        self._log_event('playbook_complete', {
            'duration_seconds': round(duration, 2),
            'hosts_count': len(stats.processed),
            'success': len([h for h in stats.processed if stats.failures.get(h, 0) == 0]),
            'failed': len([h for h in stats.processed if stats.failures.get(h, 0) > 0]),
            'timestamp': datetime.utcnow().isoformat()
        })

    def _log_event(self, event_type, data):
        log_entry = {
            'event': event_type,
            'data': data
        }
        try:
            with open(self.log_file, 'a') as f:
                f.write(json.dumps(log_entry) + '\n')
        except IOError as e:
            self._display.warning(f"Konnte nicht in {self.log_file} schreiben: {e}")

[defaults]
callback_whitelist = json_logger

---
- name: Test JSON-Logger Callback
  hosts: localhost
  gather_facts: false
  
  tasks:
    - name: Erfolgreiche Task
      debug:
        msg: "Das funktioniert"

    - name: Task mit Änderung
      copy:
        content: "Test"
        dest: /tmp/test.txt

    - name: Fehlschlagende Task (ignoriert)
      fail:
        msg: "Das ist ein Test-Fehler"
      ignore_errors: yes

# Mit Standard-Log
export ANSIBLE_JSON_LOG="/tmp/ansible_events.log"
ansible-playbook test-callback.yml

# Log anzeigen
cat /tmp/ansible_events.log | jq .

# Mit anderem Log-Pfad
ANSIBLE_JSON_LOG="/tmp/other.log" ansible-playbook test-callback.yml

mkdir -p collections/ansible_collections/company/utils/{plugins/{filter,lookup,callback},docs,tests}

# Plugins kopieren
cp plugins/filter/string_utils.py collections/ansible_collections/company/utils/plugins/filter/
cp plugins/lookup/config_file.py collections/ansible_collections/company/utils/plugins/lookup/
cp plugins/callback/json_logger.py collections/ansible_collections/company/utils/plugins/callback/

namespace: company
name: utils
version: 1.0.0
readme: README.md
authors:
  - "DevOps Team <devops@company.com>"

description: >-
  Unternehmens-spezifische Utility-Plugins:
  String-Verarbeitung, Konfiguration-Lookups, JSON-Logging

tags: [utilities, strings, config, logging]

dependencies: {}

---
- name: Test Company Utils Collection
  hosts: localhost
  gather_facts: false
  collections:
    - company.utils
    
  tasks:
    - name: String-Filter aus Collection
      debug:
        msg: "{{ 'My Service!' | to_slug }}"

    - name: Config-Lookup aus Collection
      debug:
        var: config
      vars:
        config: "{{ lookup('config_file', 'database.json') }}"

cd collections/ansible_collections/company/utils
ansible-galaxy collection build
ansible-galaxy collection install company-utils-1.0.0.tar.gz

# Testen
ansible-playbook test-collection.yml

# plugins/filter/security.py
def mask_secrets(text, patterns=None):
    """
    Ersetzt Passwörter in Text durch ***
    patterns: Liste von RegEx-Mustern für Secrets
    """
    if patterns is None:
        patterns = [r'password=\w+', r'token=\w+', r'key=\w+']
    
    # TODO: Implementierung
    # Tipp: re.sub() verwenden
    
    return masked_text

class FilterModule(object):
    def filters(self):
        return {'mask_secrets': mask_secrets}

- debug:
    msg: "{{ 'database://user:password=secret123@host' | mask_secrets }}"
# Erwartet: 'database://user:password=***@host'