🚀 Ansible Advanced: Custom Modules, Dynamic Inventory, and Complex Orchestration

Advanced Ansible: Enterprise-Level Automation

In this advanced tutorial, we’ll explore sophisticated Ansible features by implementing a complex microservices deployment with custom modules and dynamic inventory management.

Prerequisites

  • Strong understanding of Ansible basics and roles
  • Python programming experience
  • Knowledge of microservices architecture
  • Familiarity with cloud platforms (AWS/GCP/Azure)

Advanced Concepts Covered

  1. Custom Module Development
  2. Dynamic Inventory Scripts
  3. Complex Orchestration
  4. Strategy Plugins
  5. Filter Plugins
  6. Callback Plugins

Working Example: Microservices Deployment

We’ll create a sophisticated deployment that:

  • Uses custom modules for specialized tasks
  • Implements dynamic inventory for cloud resources
  • Orchestrates multiple interdependent services
  • Includes advanced monitoring and logging

1. Custom Module Development

Create a custom module for service health checks:

#!/usr/bin/python
# health_check.py

from ansible.module_utils.basic import AnsibleModule
import requests
import json

def check_service_health(url, timeout=5):
    try:
        response = requests.get(url, timeout=timeout)
        return {
            'status': response.status_code,
            'response_time': response.elapsed.total_seconds(),
            'healthy': response.status_code in range(200, 300)
        }
    except requests.RequestException as e:
        return {
            'status': None,
            'error': str(e),
            'healthy': False
        }

def main():
    module = AnsibleModule(
        argument_spec=dict(
            url=dict(required=True, type='str'),
            timeout=dict(type='int', default=5)
        )
    )

    result = check_service_health(
        module.params['url'],
        module.params['timeout']
    )

    if result['healthy']:
        module.exit_json(changed=False, **result)
    else:
        module.fail_json(msg="Service health check failed", **result)

if __name__ == '__main__':
    main()

2. Dynamic Inventory Script

Create an AWS dynamic inventory script:

#!/usr/bin/env python3
# aws_dynamic_inventory.py

import boto3
import json
import argparse

class AWSInventory:
    def __init__(self):
        self.inventory = {}
        self.ec2 = boto3.client('ec2')

    def populate(self):
        self.inventory = {
            '_meta': {'hostvars': {}},
            'all': {'children': ['ungrouped']},
            'ungrouped': {'hosts': []}
        }

        # Get all EC2 instances
        instances = self.ec2.describe_instances()
        
        for reservation in instances['Reservations']:
            for instance in reservation['Instances']:
                # Skip terminated instances
                if instance['State']['Name'] == 'terminated':
                    continue

                # Get instance details
                instance_id = instance['InstanceId']
                private_ip = instance.get('PrivateIpAddress', '')
                
                # Get tags
                tags = {t['Key']: t['Value'] for t in instance.get('Tags', [])}
                
                # Group by environment
                env = tags.get('Environment', 'development')
                if env not in self.inventory:
                    self.inventory[env] = {'hosts': []}
                self.inventory[env]['hosts'].append(instance_id)

                # Add host variables
                self.inventory['_meta']['hostvars'][instance_id] = {
                    'ansible_host': private_ip,
                    'instance_type': instance['InstanceType'],
                    'tags': tags
                }

    def json(self):
        return json.dumps(self.inventory, indent=2)

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--list', action='store_true')
    parser.add_argument('--host', action='store')
    args = parser.parse_args()

    inventory = AWSInventory()
    inventory.populate()
    print(inventory.json())

if __name__ == '__main__':
    main()

3. Complex Orchestration Playbook

Create a sophisticated deployment playbook:

---
- name: Prepare Infrastructure
  hosts: localhost
  gather_facts: false
  tasks:
    - name: Create network resources
      include_role:
        name: network_setup
      vars:
        vpc_cidr: "10.0.0.0/16"
        environment: ""

- name: Deploy Database Tier
  hosts: tag_Role_database
  become: yes
  serial: 1
  max_fail_percentage: 0
  tasks:
    - name: Include database role
      include_role:
        name: database
      vars:
        db_backup_enabled: true
        db_backup_retention: 7
      tags: ['database']

- name: Deploy Application Services
  hosts: tag_Role_application
  become: yes
  strategy: free
  tasks:
    - name: Check system resources
      include_role:
        name: system_check
      tags: ['precheck']

    - name: Deploy microservices
      include_role:
        name: microservice
      vars:
        service_name: ""
      loop:
        - auth_service
        - user_service
        - order_service
      tags: ['deploy']

    - name: Health check
      health_check:
        url: "http://:/health"
        timeout: 10
      register: health_result
      until: health_result is success
      retries: 5
      delay: 10
      tags: ['healthcheck']

- name: Configure Load Balancers
  hosts: tag_Role_loadbalancer
  become: yes
  tasks:
    - name: Update backend pool
      include_role:
        name: loadbalancer
      vars:
        backend_hosts: ""

4. Custom Filter Plugin

Create a filter for processing service configurations:

# filter_plugins/service_filters.py

class FilterModule:
    def filters(self):
        return {
            'format_service_url': self.format_service_url,
            'parse_health_check': self.parse_health_check
        }

    def format_service_url(self, hostname, port, protocol='http'):
        return f"{protocol}://{hostname}:{port}"

    def parse_health_check(self, health_result):
        if not health_result:
            return False
        
        return {
            'is_healthy': health_result.get('status', 0) in range(200, 300),
            'response_time': health_result.get('response_time', 0),
            'last_check': health_result.get('timestamp')
        }

5. Callback Plugin

Create a custom callback for Slack notifications:

# callback_plugins/slack_notify.py

from ansible.plugins.callback import CallbackBase
import requests
import json

class CallbackModule(CallbackBase):
    CALLBACK_VERSION = 2.0
    CALLBACK_TYPE = 'notification'
    CALLBACK_NAME = 'slack_notify'

    def __init__(self):
        super(CallbackModule, self).__init__()
        self.webhook_url = None

    def v2_playbook_on_start(self, playbook):
        self.webhook_url = self.get_option('webhook_url')
        self.send_notification(
            f"Starting deployment: {playbook._file_name}"
        )

    def v2_playbook_on_stats(self, stats):
        summary = self.process_stats(stats)
        self.send_notification(
            f"Deployment completed\n{json.dumps(summary, indent=2)}"
        )

    def process_stats(self, stats):
        return {
            'ok': stats.ok,
            'failures': stats.failures,
            'skipped': stats.skipped,
            'changed': stats.changed
        }

    def send_notification(self, message):
        if not self.webhook_url:
            return

        payload = {
            'text': message,
            'username': 'Ansible Deployment'
        }

        try:
            requests.post(
                self.webhook_url,
                json=payload
            )
        except Exception as e:
            self._display.warning(
                f"Failed to send Slack notification: {str(e)}"
            )

Advanced Features Demonstrated

  1. Custom Module Features:
    • Parameter validation
    • Error handling
    • Return value structure
    • Idempotency checks
  2. Dynamic Inventory Capabilities:
    • Cloud resource discovery
    • Custom grouping logic
    • Host variable management
    • Caching support
  3. Complex Orchestration:
    • Serial execution
    • Failure handling
    • Custom strategies
    • Conditional deployment

Best Practices

  1. Module Development:
    • Follow Ansible module guidelines
    • Include comprehensive documentation
    • Implement proper error handling
    • Add unit tests
  2. Inventory Management:
    • Cache results for performance
    • Handle API rate limits
    • Implement error recovery
    • Use meaningful grouping
  3. Deployment Strategy:
    • Plan for rollbacks
    • Implement circuit breakers
    • Monitor deployment health
    • Use canary deployments

Debugging and Troubleshooting

  1. Debug custom modules: 
    ANSIBLE_DEBUG=1 ansible-playbook site.yml
  2. Test dynamic inventory: 
    ./aws_dynamic_inventory.py --list
  3. Validate complex playbooks: 
    ansible-playbook site.yml --check --diff

Conclusion

You’ve now explored advanced Ansible features and learned how to implement complex automation solutions. This knowledge will help you build sophisticated, enterprise-level automation systems.

Remember to:

  • Test thoroughly in staging
  • Document custom components
  • Monitor deployment metrics
  • Plan for failure scenarios

The complete code for this tutorial is available in my GitHub repository: Advanced-Ansible-Examples

Written on July 10, 2025