基于Ansible 2 Python API实现Playbook执行重试逻辑的问询

嘿，这就给你整一个PlaybookExecutor的重试循环方案，刚好适配你已经有的自定义回调处理器！

核心思路 #

咱们的逻辑很直接：每次执行Playbook前初始化自定义回调，执行完后检查回调里的失败任务数组——如果为空，说明执行成功，直接退出；如果不为空，就触发重试，直到成功或者达到你设定的最大重试次数。

先确认你的自定义回调能正确收集失败任务 #

首先得确保你的回调处理器已经在捕获失败任务并存储，要是还没完全弄好，参考这个示例：

from ansible.plugins.callback import CallbackBase

class CustomFailureTrackingCallback(CallbackBase):
    def __init__(self):
        super().__init__()
        # 用来存所有失败的任务
        self.failed_tasks = []

    def v2_runner_on_failed(self, result, ignore_errors=False):
        # 只收集那些没被标记为ignore_errors的失败任务
        if not ignore_errors:
            self.failed_tasks.append({
                "host": result._host.get_name(),
                "task_name": result._task.get_name(),
                "error_msg": result._result.get("msg", "未返回具体错误信息")
            })

给PlaybookExecutor套上重试循环 #

接下来把你原本调用run_playbook的逻辑放进循环里，每次执行前重置回调的失败列表，执行后判断是否需要重试：

def run_playbook_with_retry(playbook_path, inventory_path, max_retries=3):
    retry_num = 0
    while retry_num < max_retries:
        # 每次重试都初始化一个新的回调实例，避免上次的失败任务残留
        callback = CustomFailureTrackingCallback()

        # 这里是你原本初始化PlaybookExecutor的代码，直接搬过来就行
        from ansible.executor.playbook_executor import PlaybookExecutor
        from ansible.inventory.manager import InventoryManager
        from ansible.vars.manager import VariableManager
        from ansible.parsing.dataloader import DataLoader

        loader = DataLoader()
        inventory = InventoryManager(loader=loader, sources=inventory_path)
        variable_manager = VariableManager(loader=loader, inventory=inventory)

        pb_executor = PlaybookExecutor(
            playbooks=[playbook_path],
            inventory=inventory,
            variable_manager=variable_manager,
            loader=loader,
            passwords={}
        )
        # 把自定义回调注册进去
        pb_executor._tqm._stdout_callback = callback

        # 执行Playbook
        exit_code = pb_executor.run()

        # 检查失败任务数组
        if not callback.failed_tasks:
            print(f"Playbook执行成功！总共重试了 {retry_num} 次")
            return exit_code
        else:
            retry_num += 1
            print(f"第 {retry_num} 次执行失败，共发现 {len(callback.failed_tasks)} 个失败任务，准备重试...")
            # 可选：打印失败任务详情，方便排查
            for task in callback.failed_tasks:
                print(f"主机 {task['host']} 的任务 {task['task_name']} 失败：{task['error_msg']}")

    # 达到最大重试次数仍失败
    print(f"已达到最大重试次数 {max_retries}，Playbook执行最终失败")
    return 1

适配你已有的嵌套调用 #

如果你的run_playbook已经嵌套在其他Python模块里，只需要把上述的循环逻辑套在原调用外面就行——或者直接把原run_playbook函数的内容替换成上面的重试逻辑，保留原函数的参数接口，这样调用方完全不用改代码就能用上重试功能。

几个实用小提示 #

• 设置合理的最大重试次数：别设太高，不然遇到永久性错误会无限循环，推荐3-5次就够了
• 重试前可加清理操作：如果某些失败任务会留下脏状态，比如临时文件、锁资源，可以在重试的else分支里加一段清理逻辑
• 区分全局重试和任务级重试：这个方案是整个Playbook失败后重试，如果你需要单个任务失败后自动重试，建议用Ansible本身的retries和delay参数，两者可以配合使用

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