Custom Plugin #
Module menjalankan task di managed node. Plugin memperluas cara kerja Ansible itu sendiri — bagaimana ia memuat data, memproses template, melaporkan output, dan menghasilkan inventory. Jika custom module adalah tentang apa yang Ansible lakukan, custom plugin adalah tentang bagaimana Ansible bekerja. Artikel ini membahas empat jenis plugin yang paling sering dibuat sendiri: lookup, filter, callback, dan inventory.
Empat Jenis Plugin yang Paling Berguna #
Lookup Plugin → Ambil data dari sumber eksternal saat playbook berjalan
Contoh: lookup('hashi_vault', ...), lookup('env', ...)
Filter Plugin → Transformasikan data dalam template Jinja2
Contoh: {{ hosts | to_yaml }}, {{ value | encrypt }}
Callback Plugin → Hook ke event lifecycle Ansible (task mulai, selesai, gagal)
Contoh: kirim notifikasi Slack saat playbook selesai
Inventory Plugin → Hasilkan inventory dinamis dari sumber eksternal
Contoh: ambil daftar server dari CMDB internal
Lookup Plugin: Ambil Data dari Sumber Eksternal #
Lookup plugin memungkinkan kamu mengambil data dari sumber apapun saat playbook berjalan — database internal, API, file khusus, atau sistem lain:
# lookup_plugins/company_cmdb.py
# Mengambil informasi server dari CMDB internal perusahaan
from ansible.plugins.lookup import LookupBase
from ansible.errors import AnsibleError
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")
cmdb_url = variables.get('cmdb_url') or kwargs.get('url', 'https://cmdb.company.internal')
cmdb_token = variables.get('cmdb_api_token', '')
headers = {"Authorization": f"Bearer {cmdb_token}"}
results = []
for term in terms:
try:
# Dukung dua format: hostname atau key=value filter
if '=' in term:
key, val = term.split('=', 1)
response = requests.get(
f"{cmdb_url}/api/servers",
params={key: val},
headers=headers,
timeout=10
)
else:
response = requests.get(
f"{cmdb_url}/api/servers/{term}",
headers=headers,
timeout=10
)
response.raise_for_status()
data = response.json()
# Normalkan ke list agar consistent
if isinstance(data, dict):
results.append(data)
else:
results.extend(data)
except requests.exceptions.ConnectionError:
raise AnsibleError(f"Tidak bisa terhubung ke CMDB di {cmdb_url}")
except requests.exceptions.HTTPError as e:
raise AnsibleError(f"CMDB mengembalikan error untuk '{term}': {e.response.status_code}")
return results
Penggunaan di playbook:
- name: Ambil informasi server dari CMDB
hosts: localhost
tasks:
- name: Dapatkan detail server tertentu
set_fact:
server_info: "{{ lookup('company_cmdb', 'web-01.company.com') }}"
- name: Dapatkan semua server dengan role database
set_fact:
db_servers: "{{ lookup('company_cmdb', 'role=database', wantlist=True) }}"
Filter Plugin: Transformasi Data dalam Template #
Filter plugin menambahkan fungsi transformasi baru yang bisa digunakan di template Jinja2 dan variabel:
# filter_plugins/company_filters.py
# Filter custom untuk kebutuhan internal
import re
import hashlib
def to_env_var(string):
"""Konversi string ke format environment variable (uppercase, underscore)."""
return re.sub(r'[^A-Z0-9]', '_', string.upper())
def mask_secret(value, visible_chars=4):
"""Tampilkan hanya beberapa karakter terakhir, sisanya disembunyikan."""
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 dan domain jika belum FQDN."""
if '.' in hostname:
return hostname
return f"{hostname}.{domain}"
def parse_size_to_bytes(size_string):
"""Konversi string ukuran (seperti '2G', '512M') ke bytes."""
units = {
'K': 1024, 'M': 1024**2, 'G': 1024**3, 'T': 1024**4
}
size_string = size_string.strip().upper()
if size_string[-1] in units:
return int(size_string[:-1]) * units[size_string[-1]]
return int(size_string)
def hash_password(password, algorithm='sha256'):
"""Hash password menggunakan algoritma yang ditentukan."""
h = hashlib.new(algorithm)
h.update(password.encode('utf-8'))
return h.hexdigest()
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,
}
Penggunaan di template dan playbook:
- name: Demonstrasi custom filters
debug:
msg:
- "{{ 'database-primary' | to_env_var }}" # DATABASE_PRIMARY
- "{{ vault_api_key | mask_secret }}" # ************abc1
- "{{ 'web-01' | server_fqdn('company.com') }}" # web-01.company.com
- "{{ '2G' | parse_size_bytes }}" # 2147483648
Callback Plugin: Hook ke Event Lifecycle #
Callback plugin memungkinkan kamu menjalankan kode custom saat event Ansible terjadi:
# callback_plugins/deployment_notifier.py
# Kirim notifikasi ke Slack saat playbook selesai
from ansible.plugins.callback import CallbackBase
import json
import time
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
DOCUMENTATION = '''
name: deployment_notifier
type: notification
short_description: Kirim notifikasi Slack saat playbook selesai
description:
- Plugin ini mengirim ringkasan hasil playbook ke Slack
options:
slack_webhook_url:
description: URL webhook Slack
env:
- name: SLACK_WEBHOOK_URL
ini:
- section: callback_deployment_notifier
key: slack_webhook_url
'''
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.stats = {}
def v2_playbook_on_start(self, playbook):
self.start_time = time.time()
self.playbook_name = playbook._file_name
def v2_playbook_on_stats(self, stats):
"""Dipanggil di akhir playbook — kirim notifikasi."""
if not HAS_REQUESTS:
return
webhook_url = self.get_option('slack_webhook_url')
if not webhook_url:
return
duration = int(time.time() - self.start_time)
hosts = sorted(stats.processed.keys())
# Hitung total statistik
total_changed = sum(stats.changed.get(h, 0) 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)
status = "✅ SUKSES"
color = "good"
if total_failures > 0 or total_unreachable > 0:
status = "❌ GAGAL"
color = "danger"
payload = {
"attachments": [{
"color": color,
"title": f"{status} — {self.playbook_name}",
"fields": [
{"title": "Host", "value": str(len(hosts)), "short": True},
{"title": "Durasi", "value": f"{duration}s", "short": True},
{"title": "Changed", "value": str(total_changed), "short": True},
{"title": "Failed", "value": str(total_failures), "short": True},
]
}]
}
try:
requests.post(webhook_url, json=payload, timeout=5)
except Exception:
pass # Jangan biarkan notifikasi yang gagal mengganggu playbook
Aktifkan di ansible.cfg:
# ansible.cfg
[defaults]
callbacks_enabled = deployment_notifier
[callback_deployment_notifier]
slack_webhook_url = https://hooks.slack.com/services/...
Lokasi Plugin #
project/
├── lookup_plugins/ # Lookup plugins untuk project ini
├── filter_plugins/ # Filter plugins
├── callback_plugins/ # Callback plugins
└── inventory_plugins/ # Inventory plugins
Untuk distribusi yang lebih luas, kemas dalam Ansible Collection:
my_namespace/my_collection/
└── plugins/
├── lookup/
├── filter/
├── callback/
└── inventory/
Ringkasan #
- Lookup plugin untuk mengambil data dari sumber eksternal (CMDB, API, database) saat playbook berjalan — hasilnya bisa digunakan langsung sebagai variabel.
- Filter plugin untuk menambahkan fungsi transformasi Jinja2 yang tidak ada bawaan — simpan dalam
FilterModule.filters()dict.- Callback plugin untuk hook ke event lifecycle Ansible — cocok untuk notifikasi otomatis, audit logging, atau integrasi dengan sistem tiket.
- Simpan plugin di direktori yang sesuai (
lookup_plugins/,filter_plugins/, dll.) di root project atau di dalam role — Ansible otomatis menemukannya.- Untuk distribusi lebih luas, kemas plugin dalam Ansible Collection — ini cara yang direkomendasikan untuk berbagi plugin antar tim.
- Selalu tangani import error dependency dengan try/except dan berikan pesan error yang informatif — plugin yang gagal karena library tidak terinstal harus memberi tahu pengguna cara mengatasinya.