Custom Plugin #

Module adalah tentang apa yang Ansible lakukan di managed node. Plugin adalah tentang bagaimana Ansible bekerja — bagaimana ia memuat data, memproses template, melaporkan output, dan menghasilkan inventory. Kalau custom module menulis kode yang dieksekusi remote, custom plugin menulis kode yang berjalan di controller dan memperluas Ansible itu sendiri: lookup untuk mengambil data eksternal saat rendering variabel, filter untuk transformasi string/struct di template Jinja2, callback untuk hook ke event lifecycle, dan inventory plugin untuk menghasilkan host list dinamis. Artikel ini membahas keempat tipe yang paling sering dibuat sendiri, lengkap dengan anatomi, best practice, dan pola anti-pattern yang harus dihindari.

Plugin vs Module: Memahami Batasannya #

Sebelum menulis plugin, pahami dulu perbedaannya. Kesalahan umum adalah menggunakan command atau shell untuk hal yang seharusnya plugin, atau sebaliknya — menggunakan module untuk hal yang sebenarnya adalah transformasi data.

flowchart LR
    subgraph "Controller"
        INV["Inventory Plugin"]
        VARS["Vars/Host Vars"]
        LP["Lookup Plugin"]
        FP["Filter Plugin"]
        CB["Callback Plugin"]
        PB["Playbook YAML"]
    end

    subgraph "Managed Node"
        MOD["Module"]
    end

    INV --> VARS
    VARS --> PB
    LP -. "inject data" .-> VARS
    FP -. "transform value" .-> VARS
    CB -. "observe events" .-> PB
    PB --> MOD
    MOD -. "result JSON" .-> PB
    PB -. "events" .-> CB

Plugin selalu berjalan di controller (lokal), module berjalan di managed node (remote). Konsekuensinya, plugin punya akses ke filesystem lokal, environment variable controller, dan API eksternal, tapi tidak bisa berinteraksi langsung dengan host yang sedang dimanage.

Tabel Perbandingan Empat Tipe Plugin #

Aspek	Lookup	Filter	Callback	Inventory
Kapan dipanggil	Saat ekspresi Jinja2 dievaluasi	Saat `\| filter_name` dipakai	Saat event Ansible terjadi	Saat `-i` diproses
Return value	List nilai (atau single value)	Nilai yang ditransformasi	Tidak ada (side effect)	Dict inventory
Use case utama	Ambil data dari API, file, DB	Format string, parse data	Logging, notifikasi, profiling	Host list dinamis dari cloud
Lokasi file	`lookup_plugins/`	`filter_plugins/`	`callback_plugins/`	`inventory_plugins/`
Akses remote	Tidak	Tidak	Tidak	Tidak
Akses ke Ansible state	Read-only variable	Read-only variable	Read-write internal state	Read variable only

Lookup Plugin: Ambil Data dari Sumber Eksternal #

Lookup plugin adalah cara Ansible untuk “menyuntikkan” data dari sumber eksternal ke variabel saat playbook berjalan. Ansible memanggilnya setiap kali ekspresi {{ lookup('nama_plugin', args) }} dievaluasi — bisa sekali, bisa ribuan kali dalam satu playbook. Karena itu, lookup yang lambat akan sangat mempengaruhi performa.

Anatomi Lookup Plugin #

sequenceDiagram
    participant PB as "Playbook Template"
    participant LP as "Lookup Plugin"
    participant EXT as "External Source"

    PB->>LP: "lookup('company_cmdb', 'web-01', wantlist=True)"
    LP->>LP: "Validasi dependencies"
    LP->>EXT: "HTTP GET /api/servers/web-01"
    EXT-->>LP: "JSON response"
    LP->>LP: "Normalisasi ke list"
    LP-->>PB: "List of dicts"
    PB->>PB: "Inject ke variable"

Implementasi Lookup Plugin #

# lookup_plugins/company_cmdb.py
# Mengambil informasi server dari CMDB internal perusahaan

from ansible.plugins.lookup import LookupBase
from ansible.errors import AnsibleError
from ansible.utils.display import Display

display = Display()

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False


class LookupModule(LookupBase):
    """
    Lookup plugin untuk CMDB internal.

    Contoh penggunaan:
      vars:
        server_info: "{{ lookup('company_cmdb', 'web-01.company.com') }}"
        all_webservers: "{{ lookup('company_cmdb', 'role=webserver', wantlist=True) }}"
    """

    def run(self, terms, variables=None, **kwargs):
        if not HAS_REQUESTS:
            raise AnsibleError(
                "Library 'requests' diperlukan untuk company_cmdb lookup plugin. "
                "Install dengan: pip install requests"
            )

        # Ambil konfigurasi dari variable global atau kwargs
        cmdb_url = (
            variables.get('cmdb_url')
            or kwargs.get('url', 'https://cmdb.company.internal')
        )
        cmdb_token = variables.get('cmdb_api_token', '')

        if not cmdb_token:
            raise AnsibleError(
                "Variable 'cmdb_api_token' belum di-set. "
                "Definisikan di group_vars/all/vault.yml."
            )

        headers = {"Authorization": f"Bearer {cmdb_token}"}
        results = []
        display.vvv(f"CMDB lookup: {len(terms)} term(s)")

        for term in terms:
            try:
                if '=' in term:
                    # Format filter: role=webserver, env=production
                    key, val = term.split('=', 1)
                    response = requests.get(
                        f"{cmdb_url}/api/servers",
                        params={key: val},
                        headers=headers,
                        timeout=10
                    )
                else:
                    # Format hostname langsung
                    response = requests.get(
                        f"{cmdb_url}/api/servers/{term}",
                        headers=headers,
                        timeout=10
                    )

                response.raise_for_status()
                data = response.json()

                # Normalisasi ke list untuk konsistensi
                if isinstance(data, dict):
                    results.append(data)
                elif isinstance(data, list):
                    results.extend(data)
                else:
                    raise AnsibleError(
                        f"CMDB mengembalikan tipe data tidak dikenal: {type(data)}"
                    )

            except requests.exceptions.ConnectionError:
                raise AnsibleError(
                    f"Tidak bisa terhubung ke CMDB di {cmdb_url}. "
                    "Periksa koneksi dan konfigurasi cmdb_url."
                )
            except requests.exceptions.HTTPError as e:
                raise AnsibleError(
                    f"CMDB mengembalikan HTTP {e.response.status_code} untuk '{term}'"
                )
            except requests.exceptions.Timeout:
                raise AnsibleError(
                    f"CMDB timeout saat memproses '{term}'. "
                    "Tingkatkan timeout atau periksa performa CMDB."
                )

        return results

Pemakaian di Playbook #

- name: "Ambil informasi server dari CMDB"
  hosts: localhost
  gather_facts: false
  tasks:
    - name: "Detail satu server"
      set_fact:
        server_info: "{{ lookup('company_cmdb', 'web-01.company.com') }}"

    - name: "Daftar semua database server"
      set_fact:
        db_servers: "{{ lookup('company_cmdb', 'role=database', wantlist=True) }}"

    - name: "Tampilkan ringkasan"
      debug:
        msg: "Database server di {{ item.datacenter }}: {{ item.ip_address }}"
      loop: "{{ db_servers }}"

Anti-Pattern: Lookup yang Melakukan Heavy I/O Berulang #

# ANTI-PATTERN: setiap term memicu HTTP request terpisah tanpa cache
def run(self, terms, variables=None, **kwargs):
    results = []
    for term in terms:
        response = requests.get(f"{API}/{term}")  # Lambat jika 100 terms
        results.append(response.json())
    return results

# BENAR: batch request atau caching internal
def run(self, terms, variables=None, **kwargs):
    # Ambil semua data sekaligus dengan satu request
    response = requests.get(f"{API}/servers", params={"ids": ",".join(terms)})
    data = {s["hostname"]: s for s in response.json()}

    results = []
    for term in terms:
        if term in data:
            results.append(data[term])
    return results

Tip — Lookup plugin dipanggil setiap kali ekspresi Jinja2 dievaluasi. Jika kita memiliki 100 host dan setiap host mengevaluasi lookup('cmdb', host.name), total ada 100 HTTP call. Untuk lookup yang mahal, pertimbangkan inventory plugin (dipanggil sekali) atau tambahkan caching internal.

Filter Plugin: Transformasi Data dalam Template #

Filter plugin menambah fungsi transformasi yang bisa dipanggil dengan sintaks {{ value | filter_name }} di Jinja2. Bedanya dengan lookup: filter menerima nilai existing dan mengembalikan nilai baru, sedangkan lookup mengambil data dari sumber eksternal. Filter murni dan idempotent — input yang sama selalu menghasilkan output yang sama.

Implementasi Filter Plugin #

# filter_plugins/company_filters.py

import re
import hashlib
import socket


def to_env_var(string):
    """Konversi string ke format environment variable (UPPER_SNAKE_CASE)."""
    return re.sub(r'[^A-Z0-9]', '_', string.upper())


def mask_secret(value, visible_chars=4):
    """Sembunyikan sebagian string, sisakan N karakter terakhir.

    Berguna untuk logging token/password tanpa mengekspos nilai penuh.
    """
    value = str(value)
    if len(value) <= visible_chars:
        return '*' * len(value)
    return '*' * (len(value) - visible_chars) + value[-visible_chars:]


def server_fqdn(hostname, domain):
    """Gabungkan hostname + domain menjadi FQDN jika belum FQDN."""
    if '.' in hostname:
        return hostname
    return f"{hostname}.{domain}"


def parse_size_to_bytes(size_string):
    """Konversi string ukuran '2G', '512M', '1024K' ke bytes."""
    units = {
        'K': 1024,
        'M': 1024 ** 2,
        'G': 1024 ** 3,
        'T': 1024 ** 4,
    }
    size_string = size_string.strip().upper()
    if not size_string:
        raise ValueError("Empty size string")

    suffix = size_string[-1]
    if suffix in units:
        return int(size_string[:-1]) * units[suffix]
    return int(size_string)


def hash_password(password, algorithm='sha256'):
    """Hash password dengan algoritma tertentu. Return hex digest."""
    if algorithm not in hashlib.algorithms_available:
        raise ValueError(f"Algorithm {algorithm} tidak tersedia")
    h = hashlib.new(algorithm)
    h.update(str(password).encode('utf-8'))
    return h.hexdigest()


def is_reachable(host, port=22, timeout=2):
    """Cek apakah host:port bisa di-connect (untuk pre-check)."""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False


class FilterModule(object):
    """Custom filters untuk kebutuhan infrastruktur perusahaan."""

    def filters(self):
        return {
            'to_env_var':       to_env_var,
            'mask_secret':      mask_secret,
            'server_fqdn':      server_fqdn,
            'parse_size_bytes': parse_size_to_bytes,
            'hash_password':    hash_password,
            'is_reachable':     is_reachable,
        }

Pemakaian di Template dan Playbook #

- name: "Demonstrasi custom filters"
  hosts: localhost
  vars:
    vault_token: "abc123supersecretvalue"
    db_host: "database-primary"
    short_host: "web-01"
    disk_quota: "10G"
  tasks:
    - name: "Generate env var name"
      debug:
        msg: "Env var: {{ db_host | to_env_var }}"
      # Output: "Env var: DATABASE_PRIMARY"

    - name: "Mask secret untuk logging"
      debug:
        msg: "Token (masked): {{ vault_token | mask_secret(6) }}"
      # Output: "Token (masked): ***************ecretvalue"

    - name: "Build FQDN"
      debug:
        msg: "Server: {{ short_host | server_fqdn('company.internal') }}"
      # Output: "Server: web-01.company.internal"

    - name: "Convert size ke bytes"
      debug:
        msg: "Quota {{ disk_quota }} = {{ disk_quota | parse_size_bytes }} bytes"
      # Output: "Quota 10G = 10737418240 bytes"

    - name: "Pre-check connectivity"
      debug:
        msg: "Host {{ item }} reachable: {{ item | is_reachable(22) }}"
      loop:
        - "web-01.internal"
        - "db-01.internal"

Anti-Pattern: Filter yang Punya Side Effect #

# ANTI-PATTERN: filter yang mengirim HTTP request
def notify_slack(message, channel='#ops'):
    """Filter 'pura-pura' tapi malah kirim notifikasi."""
    requests.post(SLACK_WEBHOOK, json={'text': message, 'channel': channel})
    return message

# BENAR: pisahkan transformasi (filter) dari side effect (callback)
def mask_secret(value, visible_chars=4):
    """Filter murni: hanya transformasi string, tidak ada side effect."""
    value = str(value)
    if len(value) <= visible_chars:
        return '*' * len(value)
    return '*' * (len(value) - visible_chars) + value[-visible_chars:]

Filter dipanggil setiap kali | filter_name muncul di template. Kalau filter punya side effect (HTTP call, write file, dsb), efek itu akan terjadi berulang tanpa kontrol. Side effect belongs in callback plugin atau task module, bukan filter.

Warning — Filter yang menulis ke file, mengirim email, atau memanggil API eksternal akan dieksekusi berkali-kali selama playbook berlangsung (sekali per ekspresi yang mereferensikannya). Ini bisa menyebabkan spam notifikasi, file corruption, atau throttle API. Filter harus murni: input X selalu menghasilkan output Y, tanpa efek samping.

Callback Plugin: Hook ke Event Lifecycle #

Callback plugin adalah plugin paling kuat untuk ekstensi Ansible. Ia menerima notifikasi untuk setiap event penting: playbook mulai, task mulai, task selesai, host selesai, playbook selesai. Pemakaian yang umum: notifikasi Slack, audit logging, profiling durasi, dan integrasi dengan sistem tiket.

Anatomi Callback Event #

sequenceDiagram
    participant EX as "Executor"
    participant CB as "Callback Plugin"
    participant EXT as "External System"

    EX->>CB: "v2_playbook_on_start(playbook)"
    Note over EX,CB: "Playbook mulai"

    loop "Per Task"
        EX->>CB: "v2_runner_on_start(task)"
        EX->>CB: "v2_runner_on_ok/task_failed/unreachable"
        Note over EX,CB: "Task result"
    end

    EX->>CB: "v2_playbook_on_handler_task_start"
    Note over EX,CB: "Handler dijalankan"

    EX->>CB: "v2_playbook_on_stats(stats)"
    Note over EX,CB: "Playbook selesai, aggregate stats"
    CB->>EXT: "Kirim notifikasi"

Tabel Event Callback yang Umum #

Event	Kapan dipanggil	Use case
`v2_playbook_on_start`	Sebelum task pertama	Set timer, log start
`v2_playbook_on_import_for_host`	Setelah import selesai	Validasi inventory
`v2_runner_on_start`	Sebelum task dieksekusi	Log task mulai
`v2_runner_on_ok`	Task sukses tanpa perubahan	Audit “ok”
`v2_runner_on_changed`	Task sukses dan mengubah state	Audit “changed”
`v2_runner_on_failed`	Task gagal	Trigger alert, rollback
`v2_runner_on_skipped`	Task di-skip	Log skip reason
`v2_runner_on_unreachable`	Host tidak reachable	Trigger PagerDuty
`v2_playbook_on_handler_task_start`	Handler dijalankan	Log handler
`v2_playbook_on_stats`	Setelah semua task selesai	Notifikasi summary

Implementasi Callback Plugin #

# callback_plugins/deployment_notifier.py
# Kirim notifikasi Slack dan audit log saat playbook selesai

from ansible.plugins.callback import CallbackBase
from ansible.utils.display import Display
import json
import time
import os

display = Display()

try:
    import requests
    HAS_REQUESTS = True
except ImportError:
    HAS_REQUESTS = False


DOCUMENTATION = '''
    name: deployment_notifier
    type: notification
    short_description: Kirim notifikasi Slack & audit log saat playbook selesai
    description:
      - Plugin ini mengirim ringkasan hasil playbook ke Slack
        dan menulis audit trail ke file JSON untuk compliance.
      - Mendukung threshold untuk alert (misal: alert jika ada >5 failed tasks).
    options:
      slack_webhook_url:
        description: URL webhook Slack
        env:
          - name: SLACK_WEBHOOK_URL
        ini:
          - section: callback_deployment_notifier
            key: slack_webhook_url
      audit_log_path:
        description: Path file audit log JSON
        default: /var/log/ansible/deployments.json
        env:
          - name: ANSIBLE_AUDIT_LOG
        ini:
          - section: callback_deployment_notifier
            key: audit_log_path
      failure_threshold:
        description: Alert jika jumlah failure melebihi threshold
        type: int
        default: 0
        env:
          - name: ANSIBLE_FAILURE_THRESHOLD
        ini:
          - section: callback_deployment_notifier
            key: failure_threshold
    requirements:
      - requests (Python library)
'''


class CallbackModule(CallbackBase):
    CALLBACK_VERSION = 2.0
    CALLBACK_TYPE = 'notification'
    CALLBACK_NAME = 'deployment_notifier'
    CALLBACK_NEEDS_ENABLED = True

    def __init__(self):
        super().__init__()
        self.start_time = None
        self.playbook_name = None
        self.task_results = []
        display.v("deployment_notifier callback initialized")

    def v2_playbook_on_start(self, playbook):
        self.start_time = time.time()
        self.playbook_name = playbook._file_name
        display.v(f"Playbook started: {self.playbook_name}")

    def v2_runner_on_ok(self, result):
        if result._result.get('changed'):
            self.task_results.append({
                'host': result._host.get_name(),
                'task': result._task.get_name(),
                'status': 'changed',
                'duration': result._result.get('delta', '0:00:00'),
            })

    def v2_runner_on_failed(self, result, ignore_errors=False):
        self.task_results.append({
            'host': result._host.get_name(),
            'task': result._task.get_name(),
            'status': 'failed',
            'error': str(result._result.get('msg', 'unknown')),
        })

    def v2_runner_on_unreachable(self, result):
        self.task_results.append({
            'host': result._host.get_name(),
            'task': result._task.get_name(),
            'status': 'unreachable',
        })

    def v2_playbook_on_stats(self, stats):
        """Dipanggil di akhir playbook — kirim notifikasi dan audit log."""
        duration = int(time.time() - self.start_time)
        hosts = sorted(stats.processed.keys())

        # Aggregate statistics
        total_changed = sum(s.get('changed', 0) for s in [
            stats.summarize(h) for h in hosts
        ])
        total_failures = sum(stats.failures.get(h, 0) for h in hosts)
        total_unreachable = sum(stats.dark.get(h, 0) for h in hosts)
        total_ok = sum(stats.ok.get(h, 0) for h in hosts)

        # Write audit log
        audit_log_path = self.get_option('audit_log_path')
        self._write_audit_log(
            audit_log_path, duration, total_changed,
            total_failures, total_unreachable, total_ok
        )

        # Send Slack notification
        if HAS_REQUESTS:
            self._send_slack_notification(
                duration, len(hosts), total_changed,
                total_failures, total_unreachable, total_ok
            )

    def _write_audit_log(self, path, duration, changed, failed, unreachable, ok):
        """Tulis audit trail ke file JSON untuk compliance."""
        audit_entry = {
            'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
            'playbook': self.playbook_name,
            'duration_seconds': duration,
            'stats': {
                'hosts': 0,
                'changed': changed,
                'failed': failed,
                'unreachable': unreachable,
                'ok': ok,
            },
            'task_results': self.task_results,
        }

        try:
            os.makedirs(os.path.dirname(path), exist_ok=True)
            with open(path, 'a') as f:
                f.write(json.dumps(audit_entry) + '\n')
        except OSError as e:
            display.warning(f"Tidak bisa tulis audit log ke {path}: {e}")

    def _send_slack_notification(self, duration, host_count, changed, failed, unreachable, ok):
        """Kirim notifikasi ke Slack dengan ringkasan hasil."""
        webhook_url = self.get_option('slack_webhook_url')
        if not webhook_url:
            return

        threshold = self.get_option('failure_threshold')
        if failed + unreachable > threshold:
            status_emoji = ":x:"
            status_text = "GAGAL"
            color = "danger"
        else:
            status_emoji = ":white_check_mark:"
            status_text = "SUKSES"
            color = "good"

        payload = {
            "attachments": [{
                "color": color,
                "title": f"{status_emoji} {status_text} — {self.playbook_name}",
                "fields": [
                    {"title": "Hosts",       "value": str(host_count),   "short": True},
                    {"title": "Duration",    "value": f"{duration}s",    "short": True},
                    {"title": "Changed",     "value": str(changed),      "short": True},
                    {"title": "Failed",      "value": str(failed),       "short": True},
                    {"title": "Unreachable", "value": str(unreachable),  "short": True},
                    {"title": "OK",          "value": str(ok),           "short": True},
                ],
                "footer": "Ansible deployment_notifier",
            }]
        }

        try:
            response = requests.post(webhook_url, json=payload, timeout=5)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            # Jangan biarkan notifikasi yang gagal mengganggu playbook
            display.warning(f"Slack notification gagal: {e}")

Konfigurasi ansible.cfg #

# ansible.cfg
[defaults]
callbacks_enabled = deployment_notifier

[callback_deployment_notifier]
slack_webhook_url = https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXX
audit_log_path = /var/log/ansible/deployments.json
failure_threshold = 0

Anti-Pattern: Callback yang Blocking #

# ANTI-PATTERN: blocking call di event hot path
def v2_runner_on_ok(self, result):
    # HTTP request sinkron di setiap task — memperlambat playbook
    requests.post(API_URL, json={'task': result._task.get_name()})

Callback plugin dipanggil sekali per task. Kalau ada 500 task dan setiap callback melakukan HTTP call 100ms, total overhead adalah 50 detik. Beberapa callback dieksekusi ratusan kali per detik untuk playbook besar.

# BENAR: aggregate dan kirim di akhir
def v2_runner_on_ok(self, result):
    # Hanya simpan ke list internal — tidak ada I/O
    self.task_results.append({'task': result._task.get_name(), 'status': 'ok'})

def v2_playbook_on_stats(self, stats):
    # Kirim satu batch summary di akhir
    summary = {'total_tasks': len(self.task_results)}
    requests.post(API_URL, json=summary, timeout=5)

Danger — Jangan pernah melempar exception dari callback plugin yang memblokir eksekusi Ansible. Callback di-handle di event hook — exception yang tidak di-handle bisa membuat playbook crash atau meninggalkan state setengah jalan. Selalu wrap side effect (HTTP call, file write) dengan try/except dan log warning.

Inventory Plugin: Hasilkan Host List Dinamis #

Inventory plugin dibahas secara mendalam di artikel Dynamic Inventory — di sini hanya disebutkan bahwa inventory plugin adalah tipe keempat dan diletakkan di direktori inventory_plugins/. Bedanya dengan inventory script (Python dengan --list/--host): plugin lebih terintegrasi dengan Ansible dan mendukung caching, keyed_groups, dan compose secara native.

# inventory_plugins/cmdb.py (sketsa, lihat artikel Dynamic Inventory untuk implementasi lengkap)
from ansible.plugins.inventory import BaseInventoryPlugin, Constructable

class InventoryModule(BaseInventoryPlugin, Constructable):
    NAME = 'cmdb'

    def verify_file(self, path):
        return path.endswith('cmdb.yml') or path.endswith('cmdb.yaml')

    def parse(self, inventory, loader, path, cache=True):
        super().parse(inventory, loader, path, cache)
        # Ambil host dari CMDB API, populasi inventory
        # Gunakan self._read_config_data untuk option
        # Gunakan self.inventory.add_host / add_group

Lokasi Plugin dan Distribusi #

Plugin bisa diletakkan di empat lokasi, masing-masing dengan cakupan berbeda:

project_root/                    # Berlaku untuk project ini saja
├── lookup_plugins/
├── filter_plugins/
├── callback_plugins/
└── inventory_plugins/

roles/
└── my_role/
    ├── lookup_plugins/         # Plugin hanya untuk role ini
    ├── filter_plugins/
    └── callback_plugins/

collections/
└── my_namespace/
    └── my_collection/
        └── plugins/
            ├── lookup/
            ├── filter/
            ├── callback/
            └── inventory/

ANSIBLE_COLLECTIONS_PATH/        # Install dari Galaxy
└── ansible_collections/
    └── my_namespace/
        └── my_collection/
            └── plugins/
                └── ...

Tip — Untuk plugin yang dipakai lintas project, kemas dalam Collection. Plugin di-root lookup_plugins/ dll. hanya ditemukan oleh Ansible saat playbook dijalankan dari direktori tersebut — ini menyulitkan sharing. Collection terinstal di path global Ansible dan bisa dipakai dari mana saja dengan FQCN.

Cara Ansible Menemukan Plugin #

flowchart TD
    A["Ansible butuh plugin 'company_cmdb'"] --> B{"Cek ANSIBLE_COLLECTIONS_PATH?"}
    B -->|Ditemukan| C["Collection: my_namespace.my_collection"]
    B -->|Tidak| D{"Cek roles/role/tipe_plugins?"}
    D -->|Ditemukan| E["Role-scoped plugin"]
    D -->|Tidak| F{"Cek tipe_plugins/ di cwd?"}
    F -->|Ditemukan| G["Project plugin"]
    F -->|Tidak| H["ERROR: plugin not found"]

Menguji Plugin #

Plugin lebih sulit diuji dibanding module karena mereka depend pada Ansible internals. Tapi testing tetap penting — bug di lookup plugin bisa menyebabkan playbook gagal dengan error cryptic.

Unit Test Filter Plugin #

# tests/test_filters.py
import sys
from unittest.mock import MagicMock

# Mock Ansible module
sys.modules['ansible'] = MagicMock()

from filter_plugins.company_filters import (
    to_env_var, mask_secret, server_fqdn, parse_size_to_bytes
)


def test_to_env_var():
    assert to_env_var("database-primary") == "DATABASE_PRIMARY"
    assert to_env_var("api-server v2") == "API_SERVER_V2"
    assert to_env_var("web.01") == "WEB_01"


def test_mask_secret():
    assert mask_secret("supersecret", 4) == "********cret"
    assert mask_secret("abc", 4) == "***"
    assert mask_secret("") == ""


def test_server_fqdn():
    assert server_fqdn("web-01", "company.com") == "web-01.company.com"
    assert server_fqdn("web-01.company.com", "company.com") == "web-01.company.com"


def test_parse_size_to_bytes():
    assert parse_size_to_bytes("2G") == 2 * 1024 ** 3
    assert parse_size_to_bytes("512M") == 512 * 1024 ** 2
    assert parse_size_to_bytes("1024") == 1024

Integration Test Callback Plugin #

# ansible.cfg
[defaults]
callbacks_enabled = deployment_notifier
stdout_callback = default

[callback_deployment_notifier]
slack_webhook_url = https://httpbin.org/post  # Test endpoint
audit_log_path = /tmp/test_audit.json

# Jalankan playbook sederhana dan verifikasi output
ansible-playbook -i localhost, -c local test_callback.yml

# Cek audit log
cat /tmp/test_audit.json | python -m json.tool

Kapan Pakai Plugin Tipe Apa #

Butuh data dari sumber eksternal saat render variabel?
  → Lookup plugin
    Contoh: ambil secret dari Vault, query database, baca file

Butuh transformasi string/struct di template Jinja2?
  → Filter plugin
    Contoh: format tanggal, parse ukuran, mask secret

Butuh hook ke event Ansible (log, notify, audit)?
  → Callback plugin
    Contoh: kirim Slack, tulis audit trail, hitung durasi

Butuh inventaris host dinamis dari cloud/CMDB?
  → Inventory plugin
    Contoh: AWS EC2, GCP, Azure, CMDB internal

Decision tree ini membantu kita memilih tipe yang tepat tanpa harus membaca semua dokumentasi Ansible. Lihat Custom Module untuk perbandingan dengan module.

Ringkasan #

Plugin memperluas cara kerja Ansible itu sendiri — selalu berjalan di controller, tidak pernah di managed node (berbeda dengan module).

Lookup plugin menarik data dari sumber eksternal saat ekspresi Jinja2 dievaluasi — hati-hati overhead karena dipanggil berulang; tambahkan batch/caching untuk data set besar.

Filter plugin melakukan transformasi murni di template — input X selalu output Y, tanpa side effect (HTTP, file write, dsb.). Side effect belongs di callback atau task.

Callback plugin hook ke event lifecycle Ansible (playbook start, task ok/failed, stats) — ideal untuk notifikasi, audit log, dan profiling; aggregate data dan kirim di akhir, jangan blocking call per event.

Simpan plugin di direktori yang sesuai (lookup_plugins/, filter_plugins/, callback_plugins/, inventory_plugins/) di root project, dalam role, atau dalam Collection untuk distribusi.

Selalu tangani import error dependency (try/except ImportError) dan berikan pesan error yang informatif — plugin yang gagal karena library tidak terinstal harus memberi tahu pengguna cara mengatasinya.

Pilih tipe plugin berdasarkan kebutuhan: lookup untuk fetch data, filter untuk transform data, callback untuk observe event, inventory untuk populate host list.

Tulis unit test untuk filter (murni, mudah diuji) dan integration test untuk lookup/callback yang berinteraksi dengan API eksternal — plugin tanpa test akan jadi sumber bug yang sulit di-debug.

Untuk distribusi lintas project, kemas dalam Collection dengan FQCN — bukan hanya diletakkan di root project, karena plugin root hanya ditemukan saat playbook dijalankan dari direktori tersebut.

← Sebelumnya: Custom Module Berikutnya: Dynamic Inventory →