Dynamic Inventory

Overview

Learn how to implement and manage dynamic inventory sources for automated host management.

Dynamic Inventory Scripts

Basic Structure

#!/usr/bin/env python3

import json

def get_inventory():
    return {
        'group1': {
            'hosts': ['host1', 'host2'],
            'vars': {
                'group_var1': 'value1'
            }
        },
        '_meta': {
            'hostvars': {
                'host1': {
                    'host_var1': 'value1'
                }
            }
        }
    }

if __name__ == '__main__':
    inventory = get_inventory()
    print(json.dumps(inventory))

Cloud Integration

AWS Example

#!/usr/bin/env python3
import boto3
import json

def aws_inventory():
    ec2 = boto3.client('ec2')
    instances = ec2.describe_instances()

    inventory = {'aws': {'hosts': []}}

    for reservation in instances['Reservations']:
        for instance in reservation['Instances']:
            if instance['State']['Name'] == 'running':
                inventory['aws']['hosts'].append(
                    instance['PublicDnsName']
                )

    return inventory

Azure Example

from azure.mgmt.compute import ComputeManagementClient
from azure.identity import DefaultAzureCredential

def azure_inventory():
    credential = DefaultAzureCredential()
    compute_client = ComputeManagementClient(
        credential,
        subscription_id
    )

Custom Inventory Sources

Database Integration

import psycopg2

def db_inventory():
    conn = psycopg2.connect(
        "dbname=inventory user=ansible"
    )
    cur = conn.cursor()
    cur.execute("SELECT hostname, groups FROM hosts")
    # Process results

API Integration

import requests

def api_inventory():
    response = requests.get(
        'https://cmdb.example.com/api/hosts',
        headers={'Authorization': 'Bearer token'}
    )
    return response.json()

Inventory Plugins

Custom Plugin Example

from ansible.plugins.inventory import BaseInventoryPlugin

class InventoryModule(BaseInventoryPlugin):
    NAME = 'custom_inventory'

    def parse(self, inventory, loader, path, cache):
        self.loader = loader
        self.inventory = inventory
        self.templar = self.loader.templar()

Caching Strategies

Redis Cache Example

import redis
import json

def cached_inventory():
    r = redis.Redis(host='localhost', port=6379, db=0)
    cached = r.get('inventory')

    if cached:
        return json.loads(cached)

    # Generate inventory
    inventory = generate_inventory()
    r.setex('inventory', 3600, json.dumps(inventory))
    return inventory

Best Practices

Error Handling

def safe_inventory():
    try:
        inventory = generate_inventory()
    except Exception as e:
        return {
            'failed': True,
            'msg': str(e)
        }
    return inventory

Performance Optimization

from concurrent.futures import ThreadPoolExecutor

def parallel_inventory():
    with ThreadPoolExecutor(max_workers=10) as executor:
        futures = [executor.submit(get_hosts) for region in regions]

Testing

Unit Tests

import unittest

class TestInventory(unittest.TestCase):
    def test_inventory_structure(self):
        inventory = get_inventory()
        self.assertIn('_meta', inventory)
        self.assertIn('hostvars', inventory['_meta'])

Integration Tests

def test_live_inventory():
    # Test against actual infrastructure
    inventory = get_inventory()
    assert len(inventory['aws']['hosts']) > 0

Troubleshooting

Common Issues

Authentication failures
API rate limiting
Stale cache data
Network connectivity

Debugging

import logging

logging.basicConfig(
    level=logging.DEBUG,
    filename='inventory.log'
)

Security Considerations

Credential Management

from ansible.parsing.vault import VaultLib

def get_credentials():
    vault = VaultLib(secret=get_vault_secret())
    return vault.decrypt(encrypted_data)

Implementation Checklist

[ ] Source identification
[ ] Authentication setup
[ ] Cache configuration
[ ] Error handling
[ ] Performance testing
[ ] Security review