Custom Plugin #

Modules are about what Ansible does on managed nodes. Plugins are about how Ansible works — how it loads data, processes templates, reports output, and generates inventory. If custom modules write code executed remotely, custom plugins write code running on the controller that extends Ansible itself: lookups for fetching external data during variable rendering, filters for string/struct transformations in Jinja2 templates, callbacks for hooking into the event lifecycle, and inventory plugins for generating dynamic host lists. This article discusses the four types most often written from scratch, complete with anatomy, best practices, and anti-patterns to avoid.

Plugin vs Module: Understanding the Boundary #

Before writing a plugin, first understand the difference. A common mistake is using command or shell for things that should be plugins, or conversely — using modules for what is actually data transformation.

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

Plugins always run on the controller (locally), modules run on the managed node (remote). Consequently, plugins have access to the local filesystem, controller environment variables, and external APIs, but can’t directly interact with the host being managed.

Comparison Table of the Four Plugin Types #

Aspect	Lookup	Filter	Callback	Inventory
When called	When a Jinja2 expression is evaluated	When `\| filter_name` is used	When an Ansible event occurs	When `-i` is processed
Return value	List of values (or a single value)	Transformed value	None (side effect)	Inventory dict
Main use case	Fetch data from APIs, files, DBs	Format strings, parse data	Logging, notifications, profiling	Dynamic host lists from the cloud
File location	`lookup_plugins/`	`filter_plugins/`	`callback_plugins/`	`inventory_plugins/`
Remote access	No	No	No	No
Access to Ansible state	Read-only variables	Read-only variables	Read-write internal state	Read variables only

Lookup Plugins: Fetching Data from External Sources #

Lookup plugins are Ansible’s way of “injecting” data from external sources into variables while the playbook runs. Ansible calls them every time the {{ lookup('plugin_name', args) }} expression is evaluated — could be once, could be thousands of times in one playbook. Because of that, a slow lookup will greatly affect performance.

Lookup Plugin Anatomy #

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: "Validate dependencies"
    LP->>EXT: "HTTP GET /api/servers/web-01"
    EXT-->>LP: "JSON response"
    LP->>LP: "Normalize to a list"
    LP-->>PB: "List of dicts"
    PB->>PB: "Inject into the variable"

Lookup Plugin Implementation #

# lookup_plugins/company_cmdb.py
# Fetches server information from the company's internal CMDB

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 for the internal CMDB.

    Example usage:
      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(
                "The 'requests' library is required for the company_cmdb lookup plugin. "
                "Install it with: pip install requests"
            )

        # Get the configuration from global variables or 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(
                "The 'cmdb_api_token' variable has not been set. "
                "Define it in 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:
                    # Filter format: 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:
                    # Direct hostname format
                    response = requests.get(
                        f"{cmdb_url}/api/servers/{term}",
                        headers=headers,
                        timeout=10
                    )

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

                # Normalize to a list for consistency
                if isinstance(data, dict):
                    results.append(data)
                elif isinstance(data, list):
                    results.extend(data)
                else:
                    raise AnsibleError(
                        f"CMDB returned an unknown data type: {type(data)}"
                    )

            except requests.exceptions.ConnectionError:
                raise AnsibleError(
                    f"Cannot connect to the CMDB at {cmdb_url}. "
                    "Check the connection and cmdb_url configuration."
                )
            except requests.exceptions.HTTPError as e:
                raise AnsibleError(
                    f"CMDB returned HTTP {e.response.status_code} for '{term}'"
                )
            except requests.exceptions.Timeout:
                raise AnsibleError(
                    f"CMDB timed out while processing '{term}'. "
                    "Increase the timeout or check CMDB performance."
                )

        return results

Usage in a Playbook #

- name: "Fetch server information from the CMDB"
  hosts: localhost
  gather_facts: false
  tasks:
    - name: "Details of one server"
      set_fact:
        server_info: "{{ lookup('company_cmdb', 'web-01.company.com') }}"

    - name: "List of all database servers"
      set_fact:
        db_servers: "{{ lookup('company_cmdb', 'role=database', wantlist=True) }}"

    - name: "Display the summary"
      debug:
        msg: "Database server in {{ item.datacenter }}: {{ item.ip_address }}"
      loop: "{{ db_servers }}"

Anti-Pattern: Lookups Doing Repeated Heavy I/O #

# ANTI-PATTERN: every term triggers a separate HTTP request without caching
def run(self, terms, variables=None, **kwargs):
    results = []
    for term in terms:
        response = requests.get(f"{API}/{term}")  # Slow if 100 terms
        results.append(response.json())
    return results

# CORRECT: batch request or internal caching
def run(self, terms, variables=None, **kwargs):
    # Fetch all data at once with a single 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 plugins are called every time a Jinja2 expression is evaluated. If we have 100 hosts and each host evaluates lookup('cmdb', host.name), that’s 100 HTTP calls total. For expensive lookups, consider an inventory plugin (called once) or add internal caching.

Filter Plugins: Data Transformation in Templates #

Filter plugins add transformation functions callable with the {{ value | filter_name }} syntax in Jinja2. The difference from lookups: filters receive an existing value and return a new value, while lookups fetch data from external sources. Filters are pure and idempotent — the same input always produces the same output.

Filter Plugin Implementation #

# filter_plugins/company_filters.py

import re
import hashlib
import socket


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


def mask_secret(value, visible_chars=4):
    """Hide part of a string, keeping the last N characters.

    Useful for logging tokens/passwords without exposing the full value.
    """
    value = str(value)
    if len(value) <= visible_chars:
        return '*' * len(value)
    return '*' * (len(value) - visible_chars) + value[-visible_chars:]


def server_fqdn(hostname, domain):
    """Combine hostname + domain into an FQDN if not already an FQDN."""
    if '.' in hostname:
        return hostname
    return f"{hostname}.{domain}"


def parse_size_to_bytes(size_string):
    """Convert a size string '2G', '512M', '1024K' to 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 a password with a specific algorithm. Return the hex digest."""
    if algorithm not in hashlib.algorithms_available:
        raise ValueError(f"Algorithm {algorithm} is not available")
    h = hashlib.new(algorithm)
    h.update(str(password).encode('utf-8'))
    return h.hexdigest()


def is_reachable(host, port=22, timeout=2):
    """Check whether host:port can be connected (for pre-checks)."""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except (socket.timeout, ConnectionRefusedError, OSError):
        return False


class FilterModule(object):
    """Custom filters for the company's infrastructure needs."""

    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,
        }

Usage in Templates and Playbooks #

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

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

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

    - name: "Convert the size to 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: Filters with Side Effects #

# ANTI-PATTERN: a filter that sends HTTP requests
def notify_slack(message, channel='#ops'):
    """A 'pretend' filter that actually sends a notification."""
    requests.post(SLACK_WEBHOOK, json={'text': message, 'channel': channel})
    return message

# CORRECT: separate transformation (filter) from side effects (callback)
def mask_secret(value, visible_chars=4):
    """A pure filter: only string transformation, no side effects."""
    value = str(value)
    if len(value) <= visible_chars:
        return '*' * len(value)
    return '*' * (len(value) - visible_chars) + value[-visible_chars:]

Filters are called every time | filter_name appears in a template. If a filter has side effects (HTTP calls, file writes, etc.), those effects happen repeatedly without control. Side effects belong in callback plugins or task modules, not filters.

Warning — A filter that writes files, sends emails, or calls external APIs will be executed many times during a playbook (once per expression referencing it). This can cause notification spam, file corruption, or API throttling. Filters must be pure: input X always produces output Y, with no side effects.

Callback Plugins: Hooking into the Event Lifecycle #

Callback plugins are the most powerful plugin type for extending Ansible. They receive notifications for every important event: playbook starts, task starts, task finishes, host finishes, playbook finishes. Common uses: Slack notifications, audit logging, duration profiling, and ticketing system integration.

Callback Event Anatomy #

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 starts"

    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 runs"

    EX->>CB: "v2_playbook_on_stats(stats)"
    Note over EX,CB: "Playbook finished, aggregate stats"
    CB->>EXT: "Send notification"

Table of Common Callback Events #

Event	When called	Use case
`v2_playbook_on_start`	Before the first task	Set a timer, log the start
`v2_playbook_on_import_for_host`	After import finishes	Validate the inventory
`v2_runner_on_start`	Before a task executes	Log task start
`v2_runner_on_ok`	Task succeeded without changes	Audit “ok”
`v2_runner_on_changed`	Task succeeded and changed state	Audit “changed”
`v2_runner_on_failed`	Task failed	Trigger an alert, rollback
`v2_runner_on_skipped`	Task skipped	Log the skip reason
`v2_runner_on_unreachable`	Host unreachable	Trigger PagerDuty
`v2_playbook_on_handler_task_start`	Handler runs	Log the handler
`v2_playbook_on_stats`	After all tasks finish	Notification summary

Callback Plugin Implementation #

# callback_plugins/deployment_notifier.py
# Sends Slack notifications and audit logs when the playbook finishes

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: Send Slack notifications & audit logs when the playbook finishes
    description:
      - This plugin sends a playbook result summary to Slack
        and writes an audit trail to a JSON file for compliance.
      - Supports an alert threshold (e.g. alert if there are >5 failed tasks).
    options:
      slack_webhook_url:
        description: The Slack webhook URL
        env:
          - name: SLACK_WEBHOOK_URL
        ini:
          - section: callback_deployment_notifier
            key: slack_webhook_url
      audit_log_path:
        description: The audit log JSON file path
        default: /var/log/ansible/deployments.json
        env:
          - name: ANSIBLE_AUDIT_LOG
        ini:
          - section: callback_deployment_notifier
            key: audit_log_path
      failure_threshold:
        description: Alert if the number of failures exceeds the 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):
        """Called at the end of the playbook — send notifications and audit logs."""
        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 the 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 the 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):
        """Write an audit trail to a JSON file for 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"Cannot write the audit log to {path}: {e}")

    def _send_slack_notification(self, duration, host_count, changed, failed, unreachable, ok):
        """Send a Slack notification with a result summary."""
        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 = "FAILED"
            color = "danger"
        else:
            status_emoji = ":white_check_mark:"
            status_text = "SUCCESS"
            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:
            # Don't let a failed notification disrupt the playbook
            display.warning(f"Slack notification failed: {e}")

ansible.cfg Configuration #

# 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: Blocking Callbacks #

# ANTI-PATTERN: blocking call on the event hot path
def v2_runner_on_ok(self, result):
    # Synchronous HTTP request on every task — slows down the playbook
    requests.post(API_URL, json={'task': result._task.get_name()})

Callback plugins are called once per task. If there are 500 tasks and every callback does a 100ms HTTP call, the total overhead is 50 seconds. Some callbacks execute hundreds of times per second for large playbooks.

# CORRECT: aggregate and send at the end
def v2_runner_on_ok(self, result):
    # Only save to an internal list — no I/O
    self.task_results.append({'task': result._task.get_name(), 'status': 'ok'})

def v2_playbook_on_stats(self, stats):
    # Send one batch summary at the end
    summary = {'total_tasks': len(self.task_results)}
    requests.post(API_URL, json=summary, timeout=5)

Danger — Never throw an exception from a callback plugin that blocks Ansible execution. Callbacks are handled in event hooks — an unhandled exception can crash the playbook or leave state half-way. Always wrap side effects (HTTP calls, file writes) with try/except and log a warning.

Inventory Plugins: Generating Dynamic Host Lists #

Inventory plugins are discussed in depth in the Dynamic Inventory article — here it’s only mentioned that the inventory plugin is the fourth type and lives in the inventory_plugins/ directory. The difference from inventory scripts (Python with --list/--host): plugins are more integrated with Ansible and natively support caching, keyed_groups, and compose.

# inventory_plugins/cmdb.py (sketch, see the Dynamic Inventory article for the full implementation)
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)
        # Fetch hosts from the CMDB API, populate the inventory
        # Use self._read_config_data for options
        # Use self.inventory.add_host / add_group

Plugin Locations and Distribution #

Plugins can be placed in four locations, each with a different scope:

project_root/                    # Applies to this project only
├── lookup_plugins/
├── filter_plugins/
├── callback_plugins/
└── inventory_plugins/

roles/
└── my_role/
    ├── lookup_plugins/         # Plugins only for this role
    ├── filter_plugins/
    └── callback_plugins/

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

ANSIBLE_COLLECTIONS_PATH/        # Installed from Galaxy
└── ansible_collections/
    └── my_namespace/
        └── my_collection/
            └── plugins/
                └── ...

Tip — For plugins used across projects, package them in a Collection. Plugins in the root lookup_plugins/ etc. are only found by Ansible when the playbook runs from that directory — this makes sharing difficult. Collections are installed in Ansible’s global path and can be used from anywhere with the FQCN.

How Ansible Finds Plugins #

flowchart TD
    A["Ansible needs the 'company_cmdb' plugin"] --> B{"Check ANSIBLE_COLLECTIONS_PATH?"}
    B -->|Found| C["Collection: my_namespace.my_collection"]
    B -->|No| D{"Check roles/role/type_plugins?"}
    D -->|Found| E["Role-scoped plugin"]
    D -->|No| F{"Check type_plugins/ in cwd?"}
    F -->|Found| G["Project plugin"]
    F -->|No| H["ERROR: plugin not found"]

Testing Plugins #

Plugins are harder to test than modules because they depend on Ansible internals. But testing is still important — a bug in a lookup plugin can make the playbook fail with a cryptic error.

Unit Testing Filter Plugins #

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

# Mock the 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 Testing Callback Plugins #

# 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

# Run a simple playbook and verify the output
ansible-playbook -i localhost, -c local test_callback.yml

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

When to Use Which Plugin Type #

Need data from an external source when rendering variables?
  → Lookup plugin
    Example: fetch a secret from Vault, query a database, read a file

Need string/struct transformation in Jinja2 templates?
  → Filter plugin
    Example: format dates, parse sizes, mask secrets

Need hooks into Ansible events (log, notify, audit)?
  → Callback plugin
    Example: send Slack, write an audit trail, measure durations

Need dynamic host inventory from the cloud/CMDB?
  → Inventory plugin
    Example: AWS EC2, GCP, Azure, internal CMDB

This decision tree helps us choose the right type without reading all of Ansible’s documentation. See Custom Module for a comparison with modules.

Summary #

Plugins extend how Ansible itself works — always running on the controller, never on the managed node (unlike modules).
Lookup plugins pull data from external sources when Jinja2 expressions are evaluated — beware the overhead because they’re called repeatedly; add batching/caching for large data sets.
Filter plugins do pure transformations in templates — input X always outputs Y, with no side effects (HTTP, file writes, etc.). Side effects belong in callbacks or tasks.
Callback plugins hook into Ansible’s event lifecycle (playbook start, task ok/failed, stats) — ideal for notifications, audit logs, and profiling; aggregate data and send at the end, don’t make blocking calls per event.
Store plugins in the right directory (lookup_plugins/, filter_plugins/, callback_plugins/, inventory_plugins/) at the project root, inside a role, or in a Collection for distribution.
Always handle dependency import errors (try/except ImportError) and give informative error messages — a plugin failing because a library isn’t installed must tell users how to fix it.
Choose the plugin type by need: lookups to fetch data, filters to transform data, callbacks to observe events, inventory to populate host lists.
Write unit tests for filters (pure, easy to test) and integration tests for lookups/callbacks interacting with external APIs — plugins without tests become a source of hard-to-debug bugs.
For cross-project distribution, package them in a Collection with the FQCN — not just placed in the project root, because root plugins are only found when the playbook runs from that directory.

← Previous: Custom Module Next: Dynamic Inventory →