如何在Python中获取Ansible Playbook执行后的不可达节点？

以下是几种无需解析命令行输出，直接通过AnsibleRunner或相关机制获取不可达节点的可靠方法：

1. 直接读取Runner的stats统计结果 #

AnsibleRunner执行完成后会自动生成执行统计数据，其中包含明确的unreachable主机列表，这是最直接且可靠的方式：

from ansible_runner import AnsibleRunner

# 初始化Runner并执行
runner = AnsibleRunner(playbook="your_playbook.yml", inventory="your_inventory.ini")
runner.run()

# 提取不可达节点
unreachable_hosts = list(runner.stats.get('unreachable', {}).keys())
print("不可达节点:", unreachable_hosts)

stats是AnsibleRunner的公开接口，稳定性有保障，无需依赖内部变量或模糊的事件解析。

2. 改进set_fact任务并提取结果 #

针对你已编写的事实注册任务，可调整为仅执行一次并通过Runner提取结果：

修改Playbook #

- name: 计算并标记不可达节点
  set_fact:
    unreachable_nodes: "{{ ansible_play_hosts_all | difference(ansible_play_hosts) }}"
  run_once: true  # 仅在控制节点执行一次，避免重复计算

- name: 输出不可达节点供Runner提取
  debug:
    var: unreachable_nodes
  run_once: true
  register: unreachable_result

通过Runner提取结果 #

执行完成后，遍历Runner的任务事件，找到对应debug任务的输出：

# 遍历事件找到目标任务结果
for event in runner.events:
    if event.get('event') == 'runner_on_ok' and event.get('task') == '输出不可达节点供Runner提取':
        unreachable_hosts = event['res']['unreachable_nodes']
        print("不可达节点:", unreachable_hosts)
        break

3. 用事件处理器精准捕获不可达事件 #

如果倾向于事件监听方式，可针对runner_on_unreachable事件做专门处理，比模糊查找事件更可靠：

from ansible_runner import AnsibleRunner

unreachable_hosts = []

def unreachable_handler(event):
    # 仅处理节点不可达的事件
    if event['event'] == 'runner_on_unreachable':
        unreachable_hosts.append(event['host'])

# 初始化Runner时传入自定义事件处理器
runner = AnsibleRunner(playbook="your_playbook.yml", inventory="your_inventory.ini", event_handler=unreachable_handler)
runner.run()

print("不可达节点:", unreachable_hosts)

关于Ansible内部API的说明 #

你提到的host_unreachable属于Ansible未公开的内部API，存在兼容性风险，官方不建议外部代码直接依赖这类接口，因此优先使用上述三种基于AnsibleRunner公开能力的方案更稳妥。

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