先说一个我自己的场景。公司内部有个批量改配置的需求业务方没有服务器登录权限只能通过运维平台点按钮。一开始我是手动敲ansible-playbook后来需求多到每天几十次我开始在 Python 里调用 Ansible API 来执行 Playbook把整个流程接进平台。这篇文章就是那段时间的实践整理从环境准备到结果解析再到几个我踩过的坑适合想用 Python 编程方式驱动 Ansible 的运维、开发和 SRE 朋友参考。网上关于 ansible-playbook 命令行的教程非常多但真正讲透 Python 调用 Ansible API 执行 Playbook 的并不多这里踩过的坑、绕过的弯路应该能帮你省不少时间。1. 为什么绕不开 API命令行之外的三个真实场景1.1 命令行面前的三道坎大部分人用 Ansible 都是从命令行开始的ansible-playbook -i hosts site.yml回车就跑起来了。这种用法在个人电脑上没有任何问题但一旦进入团队协作、平台化、自动化的场景命令行会立刻暴露三个短板。第一道坎是权限和审计。业务方不可能人手一份 SSH 私钥也不可能直接登录跳板机。如果运维把 playbook 跑起来之后还得人工把结果贴到群里一旦操作失败整个链路就没有可追溯性。这时候我们需要的是一个能被其他系统调用的入口而命令行本身不是。第二道坎是结果的结构化。命令行的 stdout 是给人看的不是给程序用的。就算你加了-v、-vv输出里也混着大量的 ANSI 颜色码和缩进。平台要做“哪些主机成功、哪些失败、失败原因是什么”的统计总不能去解析带颜色的文本。你需要的是一个能直接拿到结构化数据的接口。第三道坎是联动。比如业务方在界面上点了“发布 v1.2.3”后端要拉起一批机器、更新 inventory、传 extravars、执行 playbook再把结果回写到工单系统。这一串动作没法靠人工敲键盘完成必须由代码来驱动。所以结论很直接一旦 playbook 需要被 Web 平台、定时任务或别的程序驱动执行就必须走 API 路线。1.2 两条 API 路线的选择先别急着写代码得搞清楚“Ansible API”到底指什么。网上搜“Ansible API”搜出来的结果有时候是 AWXAnsible 的开源管理平台的 REST API有时候是ansible-runner还有人直接去 new 一个PlaybookExecutor。这三者的关系我用一句话概括AWX 的 REST API 是给平台远程调用用的ansible-runner 是官方推荐给 Python 程序本地调用用的PlaybookExecutor 是最底层、最灵活的调用方式但也最容易踩版本坑。下表是我在实际选型时基于常见实践做的一个对比方案适用场景上手难度认证方式依赖ansible-runner.run()多数自建平台、脚本集成低无需 API Keypip install ansible-runnerPlaybookExecutor需要深度定制回调、插件、上下文高无需 API Key随 ansible-core 安装AWX/Tower REST API已有管理平台、需远程编排中Token/API Key走 HTTP如果你和我一样只是想“在自己的 Python 服务里把 playbook 跑起来、拿到结果”优先选ansible-runner。它包装了临时目录的生成、inventory 的解析、事件回调、结果汇总省掉很多重复轮子。如果你还要做大规模的调度编排再考虑 AWX 的 REST API。至于 PlaybookExecutor不是说不能用而是它从 ansible-core 2.10 开始参数签名改了很多老教程的代码容易跑不通后面我会单独分析。1.3 先搞清楚一个前提API 调用不等于远程调用很多人在这一步会有一个误区以为用 Python 调用 Ansible API 就是连到一台服务器去远程执行。其实ansible-runner做的事本质上还是在你当前这台机器上执行ansible-playbook命令只是它把这套过程封装成了 Python 函数并且把输入输出做成标准化的目录和事件流。所以你的机器上必须已经装好 Ansible 本体SSH 能连到目标主机Python 解释器能加载ansible_runner模块。如果你最终需要的是一个 Web 化、多租户、可远程调用的平台那才需要 AWX 那套东西。把这两件事分开后面就不会乱。2. 环境准备与 Runner 的核心设计2.1 最小安装与版本组合安装命令很简单python3 -m venv /opt/venv source /opt/venv/bin/activate pip install --upgrade pip pip install ansible2.9 ansible-runner一般建议把 Ansible 和 ansible-runner 装进同一个 venv保证它们看到的是同一套解释器和配置目录。如果你公司内部已经有一套固定的 Ansible 版本比如走的 ansible-core 2.14那最好不要直接 pip 装最新版而是先在干净环境里验证版本兼容性。实测下来Python 3.8 以上搭配 ansible 2.9 到 2.16 这个区间基本都很稳。Python 3.6 以下建议不要用了有些依赖的语法和类型标注会出问题。另外我见过一种情况机器上有系统自带的 ansible又装了 pip 的 ansible两个版本互相干扰调用时runner找到了一个 ansibleansible-playbook命令找到的却是另一个。排查起来非常头疼。所以我的建议是尽量用 venv 隔离然后which ansible-playbook、python -c import ansible; print(ansible.__version__)两条命令先确认路径一致。2.2 private_data_dirRunner 的核心目录结构ansible-runner和直接执行命令最大的区别就是它围绕一个叫private_data_dir的目录来组织一次执行。这个目录里可以放 inventory、playbook、extravars、环境变量执行完成后还会生成 artifacts。一个典型的目录结构是/opt/runner-data/demo/ ├── inventory/ │ └── hosts # 主机清单 ├── env/ │ ├── extravars # 额外变量keyvalue 格式 │ └── envvars # 传给 ansible-playbook 的环境变量 ├── project/ │ └── site.yml # playbook 文件 └── artifacts/ └── 1700000000_1/ # 每次执行一个带时间戳的目录 ├── command # 实际执行的完整命令 ├── stdout # 完整输出 ├── rc # 退出码 ├── status # successful / failed └── events/ # 每个任务的结构化事件 JSON这段结构你不用背我当初也没刻意记但理解它之后有两个好处。第一你在run()调用时传的参数最终都会落到这些文件里第二如果你看到artifacts目录下没有东西说明你的调用参数根本没生效。排查时先去这个目录看command文件里面写的就是 runner 实际帮你拼出来的命令一眼能看出 inventory、playbook 传对了没有。2.3 为什么从 Runner 入手而不是 PlaybookExecutor社区里还有一类旧教程直接走from ansible.executor.playbook_executor import PlaybookExecutor。这条路能做但我不建议新手从这里开始。PlaybookExecutor 在 ansible-core 2.10 前后经历了一次比较大的 API 调整早期教程里的PlaybookExecutor(playbook, inventory, variable_manager, loader, passwords)这种签名在新版本已经不完全适用了。很多参数要从ansible.parsing.dataloader、ansible.vars.manager、ansible.inventory.manager里现取还要自己折腾context._init_global_context()一套初始化代码上百行最后对新手还没有任何容错。而ansible_runner.run()把这套初始化全部封装起来了你在大多数场景下只需要关注“输入什么、输出什么”。只有当你需要修改 Ansible 内部解析逻辑或者要在 playbook 执行前注入自定义的 inventory 插件时才值得去碰 PlaybookExecutor。多数团队的常见做法就是用 Runner 做集成PlaybookExecutor 只是偶尔用来看源码时才碰一下。3. 核心调用链路五步跑通第一个 Playbook3.1 最小调用一个run()跑通全部先给一个最小的可运行示例。假设你已经在/opt/runner-data/demo/project/下放了一个site.ymlimport ansible_runner result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorylocalhost,, ) print(result.rc) print(result.status) print(result.stdout)注意这里inventorylocalhost,字符串后面带一个逗号这表示一个只有 localhost 的临时清单不读文件。这样可以在本地快速验证 runner 是否正常运转因为 localhost 不需要 SSH。如果这段跑通了控制台会打印出 ansible-playbook 的完整输出result.rc是 0result.status是successful。此时你在private_data_dir/artifacts/下就能看到本次执行的stdout和rc文件。3.2 把 inventory 和变量传进去日常使用中不会只有 localhost你的 playbook 需要面对多台主机。inventory 可以是一个文件路径也可以是一个清单目录result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventory/opt/runner-data/demo/inventory/hosts, )如果你的 inventory 里定义了 group_vars、host_varsrunner 会自动去对应目录下加载不需要额外设置。变量传递是我要重点说的地方。run()的extravars参数对应命令行里的--extra-vars。标量直接写成 dict 就行result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, extravars{ app_version: 1.2.3, target_env: prod, }, )playbook 里直接{{ app_version }}就能引用。但如果你要传的是嵌套的 dict、list简单 KV 方式可能会遇到问题。因为 runner 会把 extravars 写入env/extravarsansible-playbook 再用keyvalue的方式读复杂结构在转换时容易变形。我的做法是先把复杂对象转成 JSON 字符串再传import json config { port: 8080, enabled: True, tags: [web, prod], } result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, extravars{ app_config: json.dumps(config), }, )playbook 里再用过滤器反解- set_fact: cfg: {{ app_config | from_json }}这样嵌套结构在传输过程中就不会被压扁我实际用过很多次比直接用{{ app_config[port] }}传多层 dict 稳定得多。3.3 控制主机范围、并发和超时跑生产环境前你往往需要先小范围试点再去全部机器。对应命令行的--limitRunner 里就是limit参数result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, limitapp:prod, # 只跑 prod 组里的 app 组 forks5, )forks控制并发数一般不建议给太大几十台机器并发 5 到 10 就够用。并发过大的时候SSH 连接数、被控端负载都会上来出了问题反而更难排查。Runner 还提供了一个timeout参数以秒为单位限制整个 playbook 的执行时间。如果超过时间runner 会杀掉任务result.status会变成timeoutresult ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, timeout300, )这里要区分两个概念playbook 里单个 task 的timeout是任务自己的超时而 runner 的timeout是整个执行流程的超时。我一般两个都会设任务级超时避免单步卡死整体超时兜底防止某台机器网络半通状态导致整体挂住。3.4 用 event_handler 实时观察执行过程如果你想实时拿到每一步的结果而不是等都结束再慢慢看stdout那就用到event_handler了。Runner 在执行过程中会抛出大量结构化事件event_handler就是你的观察窗口import ansible_runner def on_event(event): if event.get(event) runner_on_failed: data event.get(event_data, {}) print(f任务失败: {data.get(task)} 主机: {data.get(host)}) result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, event_handleron_event, )这个on_event会在每个事件发生时被调用事件类型里常见的有runner_on_start、runner_on_ok、runner_on_failed、runner_on_skipped、playbook_on_task_start等。如果你在做一个 Web 平台这个回调非常适合用来实时推送执行进度到前端。一个需要留意的点是回调里尽量别做耗时操作。我见过有同事在event_handler里直接写数据库执行几百个 task 时明显变慢。正确做法是把事件先丢进队列再由另一个线程写库或推 WebSocket。3.5 拿到结果对象run()返回的是一个RunnerResult对象下面这几个字段是我高频使用的字段类型含义result.rcint退出码0 表示成功result.statusstrsuccessful/failed/timeout/canceledresult.statsdict按主机聚合的 ok/failures/changed/skipped 统计result.stdoutstr完整控制台输出适合存日志result.artifacts_dirstr本次 artifact 目录路径其中result.stats长这样{ ok: {host01: 5, host02: 5}, failures: {host03: 1}, dark: {}, changed: {host01: 2}, skipped: {}, }有了它你就可以在平台界面画出“哪台机器成功、哪台失败”的统计图。失败主机在failures里无法连接的主机在dark里这两类要分开处理。提示result.rc 0不代表所有主机都成功。如果 playbook 里某些 task 设置了ignore_errors: truerc 照样是 0但stats里的failures可能不是空。做平台统计时一定以stats为准别只看 rc。4. 执行结果的正确打开方式4.1 三个结果对象各管什么Runner 给结果做了分层我平时主要用三层第一层是status和rc判断整个执行是成功还是失败第二层是stats判断具体主机层面的 ok/failure第三层是events判断每个 task 层面的执行细节。如果你需要非常细的排查比如某个 task 在某个 host 上返回了什么具体输出那就去看result.events。它是一份按时间排序的事件字典每条都对应一次回调。或者直接去artifacts_dir/events/目录下里面有完整的 JSON 文件适合事后审计。我自己的习惯是平台首页只显示stats点击单台主机再看对应 host 的 task 列表点单个 task 再从events里拉那条记录的res字段。这样从粗到细每一层都有据可查。for event_id, event in result.events.items(): if event.get(event) ! runner_on_ok: continue data event.get(event_data, {}) if data.get(host) host01: print(data.get(task), data.get(res, {}).get(stdout))res里才是模块真正的返回内容比如command模块的stdout、copy模块的dest等。4.2 事件流里的关键事件Runner 的事件类型映射着 Ansible 的官方回调掌握这几种事件类型基本就能监控整个执行生命周期事件类型触发时机常用字段playbook_on_startplaybook 开始playbook_nameplaybook_on_play_startplay 开始playplaybook_on_task_start任务开始taskrunner_on_ok任务成功host,task,resrunner_on_failed任务失败host,task,resrunner_on_unreachable主机不可达host,task,resrunner_on_skipped任务被跳过host,taskplaybook_on_statsplaybook 结束stats用event_handler时判断“最终成没成功”不要只看有没有runner_on_failed还要看有没有runner_on_unreachable。有的主机是 SSH 不通这种错误不会触发failed而是触发unreachable在统计时会被归到dark里。我第一次做平台时只拦截了runner_on_failed结果主机网络断了平台显示成功排查了半天才发现问题很尴尬。4.3 解析与编码的两个坑第一个坑是 ANIS 颜色码。如果你直接用正则去解析result.stdout里边的\x1b[0;32m一类的颜色码会干扰判断。Runner 本身不强制去掉颜色码官方没有开启--no-color的直接参数。我的做法是在 playbook 层面设置环境变量或者在调用 runner 之前先改ANSIBLE_NOCOLORimport os os.environ[ANSIBLE_NOCOLOR] True os.environ[ANSIBLE_FORCE_COLOR] False再调用ansible_runner.run()输出里就没有颜色码了写日志、做文本匹配都干净很多。第二个坑是中文乱码。playbook 里一旦有中文字符串输出result.stdout在有的 Python 版本或者终端编码环境下会出现乱码。排查时先确认一个点你传的env/extravars、inventory 文件是否都是 UTF-8 编码。Runner 本身默认按 UTF-8 处理出乱码往往是系统 locale 不对。在调用前可以export LC_ALLen_US.UTF-8或者在 Python 启动时设置locale。这个问题的表现是“命令行执行正常Python 调用后 stdout 乱码”非常像随机 bug实际上就是环境变量继承的问题。5. 现场排障从报错反推原因5.1 “module result deserialization failed”的完整排查链路先把这个报错完整写出来因为在网络搜索里它经常出现module result deserialization failed: no start of json char found完整报错通常是module result deserialization failed: no start of json char found后面可能还会带一句类似while processing module output from XXXX的信息。这个错误的本质是Ansible 通过某种连接方式SSH、local、winrm执行模块脚本后期望脚本往 stdout 里输出一段合法 JSON结果它拿到了一段不以{开头的垃圾数据无法反序列化。按我的经验排查链路分成四步。第一步先缩小范围。用命令行对同一台主机、同一个模块跑一遍ansible host01 -m shell -a echo ok如果命令行也报同样错误说明问题在模块或目标机环境本身跟 Runner 没有关系。如果命令行正常、Runner 调用报错才进入下一步。第二步检查自定义模块的 stdout。最常见的原因是模块脚本里出现了额外的print输出。比如你自己写了一个自定义模块#!/usr/bin/python import json import sys print(开始执行...) # 这行会污染 stdout result {changed: True, msg: done} print(json.dumps(result))Ansible 最终还是能识别出 JSON 部分但如果你用了sys.exit(出错了)Python 会把字符串写到 stderr而某些连接模式下 stderr 和 stdout 会被合并结果就拿不到合法的 JSON 了。正确做法是模块里所有调试信息一律写到 stdin 的辅助字段或 stderrstdout 只输出 JSON。第三步检查模块 shebang 对应的 Python 解释器是否存在。目标机上如果 shebang 写的是/usr/bin/python但那个路径下只有 Python 3模块执行时会把 traceback 混进 stdout也会造成这个错误。第四步看 artifact 目录里的 events 文件。Runner 会把模块输出原始内容记录在事件里找到那条失败事件看res里的module_stderr和module_stdout字段能直接看到目标机返回的原始内容。很多时候答案就藏在这里比如某个 shell 命令回显了 banner、某个 profile 文件打印了欢迎语。5.2 命令行能跑但 Runner 调用跑不通的三个原因这类问题非常常见表现形式是你在终端手动执行ansible-playbook没问题一到 Python 调用就报“ansible-playbook not found”“Permission denied”“host unreachable”但排障半小时也找不到原因。第一个原因是 PATH 环境变量。Runner 执行 Ansible 时如果它继承的环境里没有/opt/venv/bin它会找不到ansible-playbook。解决办法是在调用前显式传递环境变量import ansible_runner result ansible_runner.run( private_data_dir/opt/runner-data/demo, playbooksite.yml, inventorytest_hosts, envvars{ PATH: /opt/venv/bin:/usr/bin:/bin, HOME: /home/ops, }, )把 venv 路径放到 PATH 最前面能解决一半的“找不到命令”问题。第二个原因是 SSH 认证相关的环境变量丢失。比如SSH_AUTH_SOCK、SSH_AGENT_PID没有被继承导致本来能免密的 SSH 突然要密码。你可以打印这些变量看看必要时在envvars里补上。Runner 在执行时不会自动帮你 id_rsa它只负责执行 AnsibleSSH 环境是你要保证的。第三个原因是家目录不一致。Runner 在 systemd 服务或定时任务里跑时HOME往往是/root或服务的家目录而你的 SSH 密钥在别的用户目录下。这种情况比 PATH 更好伪装——你以为配好了密钥但 Ansible 用的根本不是那套密钥。排查顺序建议是先print(os.environ)再对比命令行环境最后用envvars补齐缺失项。5.3 分清 401 报错Runner 本地执行并不需要 API Key有时候你会搜到类似这样的报错unexpected status 401 unauthorized: incorrect api key provided如果你用的是ansible_runner.run()先想明白一件事Runner 是本地执行 ansible-playbook 的封装它根本不和任何 HTTP 服务做认证。除非你在 playbook 里通过uri模块调用了某个有鉴权的外部接口否则不可能出现 401。这报告错一般出现在两种地方一是你去调 AWX/Tower 的 REST API 时Token 不对或者过期了二是你去调某个 SaaS 化的自动化平台或别的 API 网关时密钥头传错了。排障顺序就三步先确认密钥是当前平台生成的不是从别处复制来的再看请求头里是不是Authorization: Bearer token这种常见格式最后检查密钥有没有过期或权限。把“Runner 本地执行”和“Restful API 远程调用”这两件事分开201 杂音会少很多。真正好用的是从 Runner 的事件流里定位模块返回的原始数据那才是治本的手段。6. 落地的两个方向与收尾建议6.1 落地场景一把 playbook 接入自建的运维平台如果你要做的平台是“用户点按钮后端跑 playbook前端实时看进度”大体上是这个流程后端接口收到请求带一个任务 ID比如订单号。根据任务 ID 动态生成private_data_dir避免多个任务共用一个目录。把 playbook、inventory、extravars 准备好。用ThreadPoolExecutor或Celery把ansible_runner.run()放到后台线程/任务队列里。用event_handler把事件推给前端。完成后把result.stats、result.rc落库。动态生成private_data_dir这个点很重要。Runner 不是线程安全的“单例工具”多个任务同时跑的时候如果都用同一个private_data_dirartifacts 目录会互相覆盖日志会打架。我见过最直接的问题就是两个发布任务同时跑后一个把前一个的 stdout 顶掉了。常见做法是import tempfile import uuid private_data_dir tempfile.mkdtemp(prefixfrunner-{uuid.uuid4().hex}-)任务跑完后按需保留 artifacts。如果只留最近 20 份可以用一个简单的清理脚本避免磁盘被事件文件占满。6.2 落地场景二定时任务的幂等与防重入如果你只是想把 playbook 变成每天凌晨三点的定时任务不需要平台那可以简单写一个 Python 入口然后配合 cron 或系统定时任务import ansible_runner def main(): result ansible_runner.run( private_data_dir/opt/runner-data/daily-check, playbookdaily_check.yml, inventoryprod_hosts, extravars{report_to: opsexample.com}, ) if result.status ! successful: # 发告警 pass if __name__ __main__: main()定时任务有一个隐蔽的问题重入。上一次执行还没结束下一次又被 cron 拉起来了会导致资源竞争。我的经验是加锁文件用fcntl实现简单的互斥import fcntl import sys def acquire_lock(lockfile/tmp/ansible_daily.lock): fp open(lockfile, w) try: fcntl.flock(fp, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: print(已有任务在执行退出) sys.exit(0) return fp另外playbook 本身要尽可能幂等。同一个 playbook 跑两遍结果应该一致。如果你在平台化落地第一次失败之后往往会自动重跑幂等性不好重跑就是二次故障。6.3 一个小习惯把“现场”留全最后分享一个我自己的习惯调用 Runner 之后不管成功失败我都会立刻把result.stdout、result.status、result.rc以及 event 里所有失败任务的原始输出写进日志。这样做的好处是事后排查时不用去翻 artifacts 目录日志本身就是完整的现场。如果问题比较诡异我还会把private_data_dir的完整路径记下来。Runner 的 artifacts 里保留了每次执行的 command 和原始事件那些 JSON 记录往往比 stdout 更适合做深度分析。我见过有人为了排查一个template模块的 inject 问题连续跑了十几次最后就是靠对比两次事件的res字段找出来的。你把 Runner 当成一个“带记录的执行器”来用而不是单纯的一个函数很多问题都会变得很好定位。从最小调用开始逐步把 inventory、变量、并发、超时、事件回调加进去你会发现 Python 调用 Ansible API 执行 Playbook 这件事本质上就是在“更适合程序工作的环境里复用你已经熟悉的 ansible-playbook”上手之后整套自动化链路会顺很多。