AWX / Ansible Automation Platform

Using Docker Compose (Development)

git clone https://github.com/ansible/awx.git
cd awx
make docker-compose-build
docker-compose -f tools/docker-compose/ansible/docker-compose.yml up -d

# Access at http://localhost:8043
# Default credentials: admin / password

Job Templates Configuration

Creating Job Templates via UI

Job Templates define how to execute playbooks with specific parameters.

Job Template Configuration:

• Name: Deploy Web Application
• Job Type: Run (or Check for dry-run)
• Inventory: Production Servers
• Project: MyApp Deployment
• Playbook: deploy.yml
• Credentials: SSH Key, Vault Password
• Extra Variables:

---
environment: production
app_version: latest
enable_rollback: true
max_fail_percentage: 10

Job Template Options:

• Limit: Target specific hosts (e.g., webserver[0:2])
• Verbosity: 0-4 for debugging output
• Job Tags: Run specific tagged tasks
• Skip Tags: Skip specific tagged tasks
• Timeout: Maximum execution time
• Privilege Escalation: Enable become/sudo
• Concurrent Jobs: Allow parallel execution
• Webhooks: Trigger via HTTP POST

Workflows and Approval Nodes

Visual Workflow Designer

Workflows chain multiple job templates together with conditional logic, approvals, and error handling.

Example Workflow: Production Deployment

START
  ↓
[Pre-deployment Checks]
  ↓
[Approval: Deploy to Production?]  ← Manual approval
  ↓ (Approved)
[Backup Current Version]
  ↓
[Deploy Application] ── (Success) → [Run Smoke Tests]
  ↓ (Failure)                           ↓ (Success)
[Rollback]                          [Notify Success]
  ↓                                     ↓
[Notify Failure]                    [END]
  ↓
[END]

Creating Workflows via API

#!/usr/bin/env python3
import requests

TOWER_URL = "https://tower.example.com"
TOKEN = "YOUR_API_TOKEN"
headers = {"Authorization": f"Bearer {TOKEN}"}

# Create workflow
workflow_data = {
    "name": "Production Deployment Workflow",
    "description": "Automated deployment with approvals",
    "organization": 1,
    "survey_enabled": True,
    "ask_variables_on_launch": True
}

response = requests.post(
    f"{TOWER_URL}/api/v2/workflow_job_templates/",
    json=workflow_data,
    headers=headers
)
workflow_id = response.json()['id']

# Add workflow nodes
nodes = [
    {
        "workflow_job_template": workflow_id,
        "unified_job_template": 10,  # Pre-check job template ID
        "identifier": "node_precheck"
    },
    {
        "workflow_job_template": workflow_id,
        "approval_node": True,
        "timeout": 3600,
        "identifier": "node_approval"
    },
    {
        "workflow_job_template": workflow_id,
        "unified_job_template": 15,  # Deploy job template ID
        "identifier": "node_deploy"
    }
]

for node in nodes:
    requests.post(
        f"{TOWER_URL}/api/v2/workflow_job_template_nodes/",
        json=node,
        headers=headers
    )

# Link nodes together
links = [
    {"source": "node_precheck", "target": "node_approval", "type": "success"},
    {"source": "node_approval", "target": "node_deploy", "type": "success"},
]

Surveys for Dynamic Input

Creating Survey Specifications

Surveys provide form-based input for job templates, allowing users to customize execution without editing playbooks.

Survey Example: Application Deployment

{
  "name": "Deployment Configuration",
  "description": "Specify deployment parameters",
  "spec": [
    {
      "question_name": "Environment",
      "question_description": "Target environment",
      "required": true,
      "type": "multiplechoice",
      "variable": "deploy_environment",
      "choices": ["development", "staging", "production"],
      "default": "staging"
    },
    {
      "question_name": "Application Version",
      "question_description": "Version tag to deploy",
      "required": true,
      "type": "text",
      "variable": "app_version",
      "default": "latest",
      "min": 1,
      "max": 50
    },
    {
      "question_name": "Number of Servers",
      "question_description": "How many servers to deploy",
      "required": true,
      "type": "integer",
      "variable": "server_count",
      "default": 3,
      "min": 1,
      "max": 10
    },
    {
      "question_name": "Enable Debug Mode",
      "question_description": "Run with verbose logging",
      "required": false,
      "type": "multiplechoice",
      "variable": "debug_mode",
      "choices": ["yes", "no"],
      "default": "no"
    },
    {
      "question_name": "Database Password",
      "question_description": "Password for database connection",
      "required": true,
      "type": "password",
      "variable": "db_password",
      "min": 8,
      "max": 128
    },
    {
      "question_name": "Deployment Notes",
      "question_description": "Optional deployment notes",
      "required": false,
      "type": "textarea",
      "variable": "deployment_notes",
      "default": ""
    }
  ]
}

Accessing Survey Variables in Playbooks

---
- name: Deploy application
  hosts: "{{ deploy_environment }}"
  vars:
    app_version: "{{ app_version }}"
    debug_enabled: "{{ debug_mode == 'yes' }}"

  tasks:
    - name: Deploy version {{ app_version }}
      docker_container:
        name: myapp
        image: "registry.example.com/myapp:{{ app_version }}"
        env:
          ENVIRONMENT: "{{ deploy_environment }}"
          DEBUG: "{{ debug_enabled }}"
          DB_PASSWORD: "{{ db_password }}"

    - name: Log deployment
      lineinfile:
        path: /var/log/deployments.log
        line: "{{ ansible_date_time.iso8601 }} - {{ deploy_environment }} - {{ app_version }} - {{ deployment_notes }}"

Role-Based Access Control (RBAC)

Organization and Team Structure

# Organization Hierarchy
MyCompany (Organization)
├── Teams:
│   ├── DevOps Team
│   │   └── Roles: Admin on Development/Staging inventories
│   ├── Developers Team
│   │   └── Roles: Execute on Development inventory only
│   ├── Operations Team
│   │   └── Roles: Execute on all inventories
│   └── Auditors Team
│       └── Roles: Read-only on all resources

# Permission Levels:
# - Admin: Full control
# - Execute: Can run job templates
# - Read: View-only access
# - Use: Can use credentials/inventories

Creating Teams and Assigning Permissions via API

#!/usr/bin/env python3
import requests

TOWER_URL = "https://tower.example.com"
TOKEN = "YOUR_API_TOKEN"
headers = {"Authorization": f"Bearer {TOKEN}"}

# Create team
team_data = {
    "name": "DevOps Team",
    "description": "DevOps engineers with deployment access",
    "organization": 1
}

team_response = requests.post(
    f"{TOWER_URL}/api/v2/teams/",
    json=team_data,
    headers=headers
)
team_id = team_response.json()['id']

# Add users to team
user_ids = [5, 7, 9]
for user_id in user_ids:
    requests.post(
        f"{TOWER_URL}/api/v2/teams/{team_id}/users/",
        json={"id": user_id},
        headers=headers
    )

# Grant job template execute permission
requests.post(
    f"{TOWER_URL}/api/v2/job_templates/10/access_list/",
    json={
        "team": team_id,
        "role_definition": "execute"
    },
    headers=headers
)

# Grant inventory use permission
requests.post(
    f"{TOWER_URL}/api/v2/inventories/5/access_list/",
    json={
        "team": team_id,
        "role_definition": "use"
    },
    headers=headers
)

Credentials Management

Credential Types

• Machine: SSH keys, passwords for managed hosts
• Source Control: Git credentials for project sync
• Vault: Ansible Vault passwords
• Network: Network device credentials
• Cloud: AWS, Azure, GCP credentials
• Custom: User-defined credential types

Creating Custom Credential Type

# Custom credential for API tokens
{
  "fields": [
    {
      "id": "api_endpoint",
      "label": "API Endpoint",
      "type": "string",
      "help_text": "Base URL for the API"
    },
    {
      "id": "api_token",
      "label": "API Token",
      "type": "string",
      "secret": true,
      "help_text": "Authentication token"
    }
  ],
  "injectors": {
    "env": {
      "MY_API_ENDPOINT": "{{ api_endpoint }}",
      "MY_API_TOKEN": "{{ api_token }}"
    },
    "extra_vars": {
      "api_endpoint": "{{ api_endpoint }}",
      "api_token": "{{ api_token }}"
    }
  }
}

API Automation

Comprehensive API Example

#!/usr/bin/env python3
"""
Complete AWX API automation script
"""
import requests
import time
import json

class AWXClient:
    def __init__(self, url, token):
        self.url = url
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

    def launch_job(self, template_id, extra_vars=None, limit=None):
        """Launch a job template"""
        payload = {}
        if extra_vars:
            payload['extra_vars'] = extra_vars
        if limit:
            payload['limit'] = limit

        response = requests.post(
            f"{self.url}/api/v2/job_templates/{template_id}/launch/",
            headers=self.headers,
            json=payload
        )
        return response.json()

    def wait_for_job(self, job_id, timeout=3600):
        """Wait for job completion"""
        start_time = time.time()

        while time.time() - start_time < timeout:
            response = requests.get(
                f"{self.url}/api/v2/jobs/{job_id}/",
                headers=self.headers
            )
            job_data = response.json()

            status = job_data['status']
            if status in ['successful', 'failed', 'error', 'canceled']:
                return job_data

            time.sleep(5)

        raise TimeoutError(f"Job {job_id} timed out")

    def get_job_output(self, job_id):
        """Get job stdout"""
        response = requests.get(
            f"{self.url}/api/v2/jobs/{job_id}/stdout/?format=txt",
            headers=self.headers
        )
        return response.text

    def create_inventory(self, name, organization_id):
        """Create dynamic inventory"""
        payload = {
            "name": name,
            "description": f"Auto-created inventory for {name}",
            "organization": organization_id
        }

        response = requests.post(
            f"{self.url}/api/v2/inventories/",
            headers=self.headers,
            json=payload
        )
        return response.json()

# Usage example
awx = AWXClient("https://tower.example.com", "YOUR_TOKEN")

# Launch deployment
job = awx.launch_job(
    template_id=10,
    extra_vars={"app_version": "v2.0.1"},
    limit="webservers[0:5]"
)

print(f"Job {job['id']} launched")

# Wait for completion
result = awx.wait_for_job(job['id'])

if result['status'] == 'successful':
    print("Deployment successful!")
    output = awx.get_job_output(job['id'])
    print(output)
else:
    print(f"Deployment failed: {result['job_explanation']}")

Integrations

LDAP/Active Directory Integration

# Configure LDAP authentication
# Settings → LDAP → LDAP Configuration

{
  "AUTH_LDAP_SERVER_URI": "ldap://ldap.example.com:389",
  "AUTH_LDAP_BIND_DN": "cn=ansible,ou=services,dc=example,dc=com",
  "AUTH_LDAP_BIND_PASSWORD": "password",
  "AUTH_LDAP_USER_SEARCH": [
    "ou=users,dc=example,dc=com",
    "SCOPE_SUBTREE",
    "(uid=%(user)s)"
  ],
  "AUTH_LDAP_GROUP_SEARCH": [
    "ou=groups,dc=example,dc=com",
    "SCOPE_SUBTREE",
    "(objectClass=groupOfNames)"
  ],
  "AUTH_LDAP_GROUP_TYPE": "GroupOfNamesType",
  "AUTH_LDAP_USER_FLAGS_BY_GROUP": {
    "is_superuser": "cn=admins,ou=groups,dc=example,dc=com"
  },
  "AUTH_LDAP_ORGANIZATION_MAP": {
    "MyOrg": {
      "admins": "cn=tower-admins,ou=groups,dc=example,dc=com",
      "users": "cn=tower-users,ou=groups,dc=example,dc=com"
    }
  }
}

Webhook Triggers

# GitHub Webhook Configuration
# Job Template → Options → Enable Webhook

# GitHub webhook URL:
# https://tower.example.com/api/v2/job_templates/10/github/

# Webhook secret: 

# GitHub repository settings:
# Settings → Webhooks → Add webhook
# Payload URL: https://tower.example.com/api/v2/job_templates/10/github/
# Content type: application/json
# Secret: 
# Events: Push, Pull Request

# Trigger job on git push
curl -X POST https://tower.example.com/api/v2/job_templates/10/github/ \
  -H "Content-Type: application/json" \
  -H "X-Hub-Signature: sha1=" \
  -d '{"ref":"refs/heads/main"}'

Notification Integration

# Notification Templates
# Administration → Notifications

# Slack notification:
{
  "notification_type": "slack",
  "notification_configuration": {
    "token": "xoxb-your-slack-token",
    "channels": ["#deployments"],
    "username": "Ansible Tower",
    "icon_url": "https://example.com/ansible-icon.png"
  },
  "messages": {
    "started": {
      "message": "Job {{ job_name }} started by {{ job.created_by }}"
    },
    "success": {
      "message": "✅ Job {{ job_name }} completed successfully"
    },
    "error": {
      "message": "❌ Job {{ job_name }} failed: {{ job.result_traceback }}"
    }
  }
}

# Email notification:
{
  "notification_type": "email",
  "notification_configuration": {
    "username": "ansible@example.com",
    "password": "smtp-password",
    "host": "smtp.example.com",
    "port": 587,
    "use_tls": true,
    "sender": "ansible-tower@example.com",
    "recipients": ["ops-team@example.com"]
  }
}

# PagerDuty notification:
{
  "notification_type": "pagerduty",
  "notification_configuration": {
    "token": "your-pagerduty-token",
    "subdomain": "your-subdomain",
    "service_key": "service-integration-key",
    "client_name": "Ansible Tower"
  }
}

Monitoring and Reporting

Job Analytics

# Query job statistics via API
import requests
from datetime import datetime, timedelta

TOWER_URL = "https://tower.example.com"
TOKEN = "YOUR_API_TOKEN"
headers = {"Authorization": f"Bearer {TOKEN}"}

# Get jobs from last 7 days
seven_days_ago = (datetime.now() - timedelta(days=7)).isoformat()

response = requests.get(
    f"{TOWER_URL}/api/v2/jobs/",
    headers=headers,
    params={
        "created__gte": seven_days_ago,
        "page_size": 200
    }
)

jobs = response.json()['results']

# Calculate statistics
stats = {
    'total': len(jobs),
    'successful': len([j for j in jobs if j['status'] == 'successful']),
    'failed': len([j for j in jobs if j['status'] == 'failed']),
    'avg_duration': sum([j['elapsed'] for j in jobs if j['elapsed']]) / len(jobs)
}

print(f"Success rate: {stats['successful'] / stats['total'] * 100:.2f}%")
print(f"Average duration: {stats['avg_duration']:.2f} seconds")

Best Practices Summary