基于Flask与Python实现Ansible日志实时捕获并在HTML展示的问题

我正在做一个实验，通过Flask应用执行Ansible Playbook，想借助Socket.IO把Ansible的输出日志实时捕获并推送到HTML页面展示。我已经设置了quiet=False并尝试用runner.events获取日志，但始终没法实现实时推送——日志只会在Playbook完全执行完成后才一次性发送。我对Ansible的Python API完全不熟悉，求技术帮助。

#
# Ansible code
# 
@app.route('/generate_ansible_inventory', methods=['GET'])
def generate_ansible_inventory():
    # Load data from TinyDB
    vm_data_list = load_vm_from_db()

    # Create an empty inventory dictionary
    inventory = {"all": {"hosts": {}, "vars": {}}}

    # Loop through VMs in the TinyDB
    for vm_data in vm_data_list:
        # Add an inventory host
        inventory["all"]["hosts"][vm_data["vm_name"]] = {
            "ansible_host": vm_data["vm_ipv4_address"],
            "ansible_user": "bart",
            "ansible_ssh_pass": "bart",
            "ansible_become_pass": "bart"
        }

    # Save the inventory dictionary to a json file in the Ansible directory
    with open('ansible/inventory.json', 'w') as file:
        json.dump(inventory, file, indent=4)

    return jsonify(inventory)

# Run Ansible playbook asynchronously
def run_ansible_playbook_async(playbook_path, inventory_path):
    try:
        # Read the inventory from the JSON file
        with open(inventory_path, 'r') as file:
            inventory = json.load(file)

        # Run the playbook
        runner = ansible_runner.run(
            playbook=playbook_path,
            inventory=inventory,
            quiet=False,  # Set quiet to False to capture logs
        )

        # Emit each log line to connected clients
        for event in runner.events:
            if event['event'] == 'runner_on_ok' and 'stdout' in event['stdout']:
                socketio.emit('ansible_output', {'log_message': event['stdout']['stdout_lines'][0]})
            elif event['event'] == 'runner_on_failed' and 'stderr' in event['stdout']:
                socketio.emit('ansible_output', {'log_message': event['stdout']['stdout_lines'][0]})

        # Print the output
        print("Playbook run status:", runner.status)
        print("Playbook run stats:", runner.stats)

    except Exception as e:
        print("Error running the Ansible playbook:", e)


# Route for running Ansible playbook
@app.route('/run_ansible_playbook', methods=['GET'])
def run_ansible_playbook():
    try:
        # Define playbook and inventory paths
        playbook_path = os.path.join(os.getcwd(), 'ansible', 'playbook.yml')
        inventory_path = os.path.join(os.getcwd(), 'ansible', 'inventory.json')

        # Start a new thread for running Ansible playbook asynchronously
        ansible_thread = Thread(target=run_ansible_playbook_async, args=(playbook_path, inventory_path))
        ansible_thread.start()

        return jsonify({"status": "Ansible playbook execution started in the background"})

    except Exception as e:
        return jsonify({"status": f"Error running Ansible playbook: {str(e)}"})

@app.route('/ansible_logs', methods=['GET'])
def ansible_logs():
    return render_template('ansible_logs.html')

问题根源 #

你当前的写法是先调用ansible_runner.run()等待整个Playbook执行完毕，再遍历runner.events推送日志——这本质上是事后批量推送，而非实时推送。ansible_runner.run()是阻塞式方法，只有Playbook全跑完才会返回runner对象，自然没法拿到实时事件。

修改方案：用事件回调实现实时推送 #

Ansible Runner支持通过event_handler参数传入回调函数，每产生一个事件就触发一次回调，这样就能在事件发生时立刻推送给前端。

步骤1：定义事件回调函数

添加一个专门处理Ansible事件的函数，在函数里根据事件类型提取日志并通过Socket.IO推送：

def handle_ansible_event(event_data):
    # 提取事件类型和关键信息
    event_type = event_data['event']
    log_msg = ""

    # 处理不同类型的事件
    if event_type == 'runner_on_stdout':
        # 捕获模块执行的标准输出
        log_msg = event_data['event_data']['stdout']
    elif event_type == 'runner_on_stderr':
        # 捕获模块执行的错误输出
        log_msg = event_data['event_data']['stderr']
    elif event_type == 'runner_on_ok':
        # 任务执行成功的提示
        task_name = event_data['event_data']['task']['name']
        host = event_data['event_data']['host']
        log_msg = f"[SUCCESS] {host} - {task_name}"
    elif event_type == 'runner_on_failed':
        # 任务执行失败的提示
        task_name = event_data['event_data']['task']['name']
        host = event_data['event_data']['host']
        error_msg = event_data['event_data']['res']['msg']
        log_msg = f"[FAILED] {host} - {task_name}: {error_msg}"
    elif event_type == 'playbook_on_start':
        # Playbook开始执行的提示
        log_msg = "=== Playbook 开始执行 ==="
    elif event_type == 'playbook_on_stats':
        # Playbook执行完成的统计信息
        stats = event_data['event_data']['stats']
        log_msg = f"=== 执行完成 ==> 成功: {stats['ok']}, 失败: {stats['failures']}, 跳过: {stats['skipped']}"

    # 有日志内容才推送
    if log_msg:
        socketio.emit('ansible_output', {'log_message': log_msg.strip()})

步骤2：修改Playbook执行代码

将ansible_runner.run()的调用改为传入event_handler参数，让Runner在执行过程中实时触发回调：

def run_ansible_playbook_async(playbook_path, inventory_path):
    try:
        with open(inventory_path, 'r') as file:
            inventory = json.load(file)

        # 调用runner时传入事件回调，实时处理事件
        runner = ansible_runner.run(
            playbook=playbook_path,
            inventory=inventory,
            quiet=False,
            event_handler=handle_ansible_event  # 添加事件回调
        )

        # 执行完成后输出统计（可选）
        print("Playbook run status:", runner.status)
        print("Playbook run stats:", runner.stats)

    except Exception as e:
        # 推送错误信息到前端
        socketio.emit('ansible_output', {'log_message': f"[ERROR] 执行Playbook失败: {str(e)}"})
        print("Error running the Ansible playbook:", e)

额外注意事项 #

1. 前端监听确认：确保你的ansible_logs.html里正确监听了Socket.IO的ansible_output事件，比如：

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
<script>
    var socket = io();
    socket.on('ansible_output', function(data) {
        var logDiv = document.getElementById('ansible-logs');
        logDiv.innerHTML += '<p>' + data.log_message + '</p>';
        // 自动滚动到最新日志
        logDiv.scrollTop = logDiv.scrollHeight;
    });
</script>

2. 线程上下文问题：如果使用Flask-SocketIO，在子线程中调用socketio.emit时，可能需要手动推送应用上下文，可在回调函数开头添加：

from flask import current_app

def handle_ansible_event(event_data):
    with current_app.app_context():
        # 原有的事件处理代码
        ...

3. 事件覆盖范围：上面的回调函数覆盖了常见的事件类型，你可以根据需求添加更多事件（比如runner_on_skipped、playbook_on_task_start等）。

内容的提问来源于stack exchange，提问作者Bartballon