🚀 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: "staging" - 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

1. Test dynamic inventory:

./aws_dynamic_inventory.py --list

1. 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