Ansible Advanced Techniques — Level Up Your Automation

#!/usr/bin/python
# library/app_health.py

from ansible.module_utils.basic import AnsibleModule
import requests

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

    url = module.params['url']
    timeout = module.params['timeout']
    expected = module.params['expected_status']

    try:
        resp = requests.get(url, timeout=timeout)
        if resp.status_code == expected:
            module.exit_json(
                changed=False,
                status_code=resp.status_code,
                healthy=True,
                response_time=resp.elapsed.total_seconds()
            )
        else:
            module.fail_json(
                msg=f"Expected {expected}, got {resp.status_code}",
                status_code=resp.status_code
            )
    except requests.exceptions.RequestException as e:
        module.fail_json(msg=str(e))

if __name__ == '__main__':
    main()

# Use your custom module
- name: Check application health
  app_health:
    url: "http://localhost:{{ app_port }}/health"
    timeout: 5
  register: health

- debug:
    msg: "Response time: {{ health.response_time }}s"

# filter_plugins/custom_filters.py

def to_environment(env_dict):
    """Convert dict to KEY=VALUE environment format."""
    return '\n'.join(f'{k}={v}' for k, v in env_dict.items())

def mask_password(text, mask='****'):
    """Mask password in connection strings."""
    import re
    return re.sub(r'password=[^&\s]+', f'password={mask}', text)

class FilterModule:
    def filters(self):
        return {
            'to_environment': to_environment,
            'mask_password': mask_password,
        }

- debug:
    msg: "{{ db_config | to_environment }}"
# Output: DB_HOST=localhost\nDB_PORT=5432\nDB_NAME=myapp

# Deep merge configurations
- set_fact:
    final_config: "{{ default_config | combine(env_config, recursive=True) | combine(host_config, recursive=True) }}"

# Build dynamic inventory from API
- uri:
    url: https://cmdb.example.com/api/servers
    return_content: true
  register: cmdb

- set_fact:
    web_servers: "{{ cmdb.json | selectattr('role', 'eq', 'web') | map(attribute='hostname') | list }}"
    db_servers: "{{ cmdb.json | selectattr('role', 'eq', 'db') | map(attribute='hostname') | list }}"

# Multi-condition logic
- name: Deploy only if conditions met
  ansible.builtin.include_tasks: deploy.yml
  when:
    - deploy_enabled | default(true) | bool
    - ansible_distribution_version is version('20.04', '>=')
    - inventory_hostname not in maintenance_hosts | default([])
    - "'canary' not in group_names or canary_approved | default(false)"

# Complex filtering
- set_fact:
    production_apps: "{{ apps | json_query('[?env==`production` && status==`active`].{name: name, port: port}') }}"
    
    high_memory: "{{ instances | json_query('[?memory_gb > `16`]') }}"
    
    grouped: "{{ servers | json_query('[].{region: region, hosts: hosts[?state==`running`].hostname}') }}"

# Run on localhost, loop over remote hosts
- name: Add DNS records for all hosts
  community.general.cloudflare_dns:
    zone: example.com
    record: "{{ item }}"
    type: A
    value: "{{ hostvars[item]['ansible_host'] }}"
  delegate_to: localhost
  loop: "{{ groups['webservers'] }}"
  run_once: true

# Delegate facts
- name: Get DB version
  ansible.builtin.command: psql --version
  register: db_version
  delegate_to: "{{ groups['dbservers'][0] }}"
  delegate_facts: true
  run_once: true

# Retry with exponential backoff
- name: Wait for service with backoff
  ansible.builtin.uri:
    url: "http://{{ app_host }}:{{ app_port }}/health"
  register: result
  until: result.status == 200
  retries: 10
  delay: "{{ item }}"
  loop: [1, 2, 4, 8, 16, 30, 30, 30, 30, 30]

# Block with rescue and notification
- block:
    - name: Deploy application
      include_tasks: deploy.yml
      
    - name: Verify deployment
      uri:
        url: http://localhost:{{ app_port }}/health
        status_code: 200
      retries: 10
      delay: 5

  rescue:
    - name: Rollback on failure
      include_tasks: rollback.yml

    - name: Notify team
      uri:
        url: "{{ slack_webhook }}"
        method: POST
        body_format: json
        body:
          text: "⚠️ Deploy FAILED on {{ inventory_hostname }}. Rolled back."
      delegate_to: localhost

  always:
    - name: Record deployment
      uri:
        url: https://deployments.example.com/api/record
        method: POST
        body_format: json
        body:
          host: "{{ inventory_hostname }}"
          version: "{{ app_version }}"
          status: "{{ 'failed' if ansible_failed_result is defined else 'success' }}"
      delegate_to: localhost
      ignore_errors: true

# Choose role based on variables
- name: Apply web server role
  include_role:
    name: "{{ web_server_role }}"
  vars:
    web_server_role: "{{ 'nginx' if use_nginx | default(true) else 'apache' }}"

# Role per OS
- name: Apply OS-specific role
  include_role:
    name: "{{ ansible_distribution | lower }}_setup"

# Deploy custom fact script to remote hosts
- name: Deploy custom facts
  ansible.builtin.copy:
    content: |
      #!/bin/bash
      echo '{"app_version": "{{ app_version }}", "deploy_date": "{{ ansible_date_time.iso8601 }}", "deployed_by": "ansible"}'
    dest: /etc/ansible/facts.d/myapp.fact
    mode: "0755"

# Access custom facts
- name: Show app version
  debug:
    msg: "Version: {{ ansible_local.myapp.app_version }}"

# Encrypt specific variables inline
vars:
  db_password: !vault |
    $ANSIBLE_VAULT;1.1;AES256
    613838...

# Multiple vault IDs
ansible-playbook site.yml --vault-id dev@prompt --vault-id prod@vault-pass-prod

# Vault in lookup
- name: Read encrypted file
  debug:
    msg: "{{ lookup('file', 'secrets/api_key.vault') }}"

# Flush handlers mid-play
- meta: flush_handlers

# End play for current host
- meta: end_host
  when: skip_this_host | default(false)

# End entire play
- meta: end_play
  when: critical_failure

# Refresh facts
- meta: clear_facts
- setup:    # Re-gather

# Clear host errors
- meta: clear_host_errors

# Conditional fact gathering (only what you need)
- setup:
    gather_subset:
      - '!all'
      - network
      - hardware

# Throttle API calls
- name: Register with API
  uri:
    url: https://api.example.com/register
    method: POST
    body_format: json
    body: { hostname: "{{ inventory_hostname }}" }
  throttle: 5    # Max 5 concurrent

# Pre-cache data on controller
- name: Fetch server list once
  uri:
    url: https://api.example.com/servers
  register: server_list
  run_once: true
  delegate_to: localhost

# Assert expected state
- name: Verify deployment
  ansible.builtin.assert:
    that:
      - app_health.status == 200
      - "'v{{ app_version }}' in app_health.content"
      - db_check.rc == 0
      - (disk_check.stdout | int) > 20
    fail_msg: "Post-deploy verification failed"

# Molecule with multiple scenarios
molecule test -s docker
molecule test -s vagrant
molecule test -s cloud