Ansible回调插件中如何获取playbook的extra arguments及实现runID标记计时

阿华AIGC实验室

2026-5-20

获取Ansible Callback Plugin中的Playbook额外参数（runID）

我刚好碰到过类似的需求，给你梳理下可行的解决方案，核心在于找对Callback Plugin的钩子方法，以及正确获取额外参数的路径：

1. 关键：在Playbook启动时捕获runID

你需要在v2_playbook_on_start钩子中获取传入的runID——这个钩子会在整个Playbook开始执行时触发，能直接访问到Playbook级别的额外参数。

示例代码片段：

from ansible.plugins.callback import CallbackPlugin
import datetime
import uuid

class CallbackModule(CallbackPlugin):
    CALLBACK_VERSION = 2.0
    CALLBACK_NAME = 'task_timer'
    CALLBACK_NEEDS_WHITELIST = True  # Ansible 2.10+版本可省略

    def __init__(self):
        super().__init__()
        self.run_id = None
        self.task_start_times = {}  # 存储每个task的开始时间

    def v2_playbook_on_start(self, playbook):
        # 从extra_vars中提取runID
        self.run_id = playbook.extra_vars.get('runID')
        # 兜底：如果没有传入runID，生成一个默认唯一ID
        if not self.run_id:
            self.run_id = f"default-run-{str(uuid.uuid4())}"

    # 记录task开始时间
    def v2_runner_on_start(self, host, task):
        self.task_start_times[task._uuid] = datetime.datetime.now()

    # 统一处理不同状态的task结果
    def v2_runner_on_ok(self, result):
        self._process_task_result(result, 'ok')

    def v2_runner_on_failed(self, result, ignore_errors=False):
        self._process_task_result(result, 'failed')

    def v2_runner_on_skipped(self, result):
        self._process_task_result(result, 'skipped')

    def _process_task_result(self, result, status):
        start_time = self.task_start_times.pop(result._task._uuid, None)
        if not start_time:
            return
        
        # 计算耗时（保留两位小数）
        duration = (datetime.datetime.now() - start_time).total_seconds()
        # 组装上报数据，带上runID标记
        log_payload = {
            'runID': self.run_id,
            'task_name': result._task.get_name().strip(),
            'host': result._host.get_name(),
            'duration_seconds': round(duration, 2),
            'status': status,
            'task_uuid': result._task._uuid
        }

        # 替换成你的logging agent上报逻辑（比如HTTP请求、消息队列等）
        self._send_to_logging_agent(log_payload)

    def _send_to_logging_agent(self, payload):
        # 示例：打印到控制台，实际替换成你的上报代码
        print(f"[Task Timer] Reporting: {payload}")

2. 验证与注意事项

传入runID的方式：确保你用--extra-vars（或-e）参数传入，比如：
```
ansible-playbook -e "runID=prod-deploy-20240520-001" your_deploy_playbook.yml
```
Callback Plugin的加载：把这个插件文件放在callback_plugins目录下（或在ansible.cfg中配置callback_plugins路径），然后启用它：
```
[defaults]
# Ansible 2.10+用callback_enabled，旧版本用callback_whitelist
callback_enabled = task_timer
```
多Play场景：如果你的Playbook包含多个Play，runID会全局生效；如果需要在单个Play级别覆盖，可在v2_play_on_start钩子中重新读取Play的vars更新self.run_id。