如何基于 ansible-api 二次开发?

作者: 王炳明 分类: Linux 教程 发布时间: 2020-12-27 14:07 热度:383

长久以来,IT 运维在企业内部一直是个耗人耗力的事情。随着虚拟化的大量应用、私有云、容器的不断普及,数据中心内部的压力愈发增加。传统的自动化工具,往往是面向于数据中心特定的一类对象,例如操作系统、虚拟化、网络设备的自动化运维工具往往是不同的。那么,有没有一种数据中心级别的统一的自动化运维工具呢?

答案就是 Ansible。和传统的自动化工具 (如 Puppet)相比,ansible 尤其明显的优势:

  • 简单,是一种高级的脚本类语言,而非标准语言。
  • 不需要安装 agent, 分为管理节点和远程被管节点,通过 SSH 认证。
  • 纳管范围广泛,不仅仅是操作系统,还包括各种虚拟化、公有云,甚至网络设备。

用过 ansible 的人都知道,ansible 一般都是通过命令行执行的:ansible、ansible-playbook、ansible-doc、ansible-galaxy、ansible-console 等。用执行命令的方式,整个过程的所有信息都会实时的打印在终端屏幕。这样的好处是,反馈实时,可以观察整个部署过程。缺点就是信息过多,难以获取有效的信息。

这时候,我们自然想到可以直接调用 ansible 提供的接口进行开发。

通过搜索引擎查询有关 ansible 二次开发的相关资料,发现这方面的文章非常少,而且由于 ansible 一直在进行代码的重构优化,有很多网友提供的 demo 已不适用我当前使用的版本,一直跑不通。

本想从官方文档中查得一些资料,结果发现官网中对于这块也没有很详细的介绍。

后来想想,ansible、ansible-playbook、ansible-doc 这些命令不也是基于 ansible 的 api 开发出来的吗?

我不是可以模仿他们,来调用 ansible api 接口。

以下通过阅读 ansible cli 的源码,总结出来的。

1. 如何调用 Playbook

我习惯性地将调用 playbook 的代码封装为一个函数。

def playbook_task():
    try:
        # 执行 playbook 的各项参数
        cli = PlaybookCLI(args=["", "you_task.yml"])
        cli.parse()
        super(PlaybookCLI, cli).run()

        loader, inventory, variable_manager = cli._play_prereqs(cli.options)
        CLI.get_host_list(inventory, cli.options.subset)
        pbex = PlaybookExecutor(playbooks=cli.args, inventory=inventory,
                                variable_manager=variable_manager, 
                                loader=loader, 
                                options=cli.options,
                                passwords=None)
        # 【重点】指定回调实例
        pbex._tqm._stdout_callback = ResultCallback()
        exit_code = pbex.run()

    except Exception as e:
        LOG.exception('Failed to run playbook:{}'.format(e))

以上的代码相对固定,你可以直接拷贝直接用。除了两个位置你要注意一下

  1. 第一个注释处:按需要对应添加 playbook 的参数
  2. 第二个注释处:指定你自定义的回调实例

定义回调类

from ansible.plugins.callback.default import CallbackModule

class ResultCallback(CallbackModule):
    def __init__(self):
        self.display_failed_stderr = False
        self.display_ok_hosts = True
        super(ResultCallback, self).__init__()

    def v2_runner_on_failed(self, result, ignore_errors=False, **kwargs):
        super(ResultCallback, self).v2_runner_on_failed(result, ignore_errors, **kwargs)
        host = result._host.get_name()
        self.runner_on_failed(host, result._result, ignore_errors)
        if not ignore_errors:
            if "ws" not in host:
                host = "ws_" + host.replace("_host", "")

            if host in task_result:
                task_result[host]["detailStatus"] = "FAIL"
                task_result[host]["detailMsg"] = result.task_name + ": " + result._result.get("msg", "").encode("utf-8")
            else:
                for host, _ in task_result.items():
                    task_result[host]["detailStatus"] = "FAIL"
                    task_result[host]["detailMsg"] = result.task_name + ": " + result._result.get("msg", "").encode("utf-8")

    def v2_runner_on_unreachable(self, result):
        super(ResultCallback, self).v2_runner_on_unreachable(result)
        host = result._host.get_name()
        self.runner_on_unreachable(host, result._result)
        if "ws" not in host:
            host = "ws_" + host.replace("_host", "")
        task_result[host]["detailStatus"] = "FAIL"
        task_result[host]["detailMsg"] = result.task_name + ": " + result._result.get("msg", "").encode("utf-8")

    def v2_runner_on_skipped(self, result):
        super(ResultCallback, self).v2_runner_on_skipped(result)
        if C.DISPLAY_SKIPPED_HOSTS:
            host = result._host.get_name()
            self.runner_on_skipped(host, self._get_item(getattr(result._result, 'results', {})))

并实现其中几个关键的方法

  • v2_runner_on_ok:单个部署任务成功时,会调用
  • v2_runner_on_failed:节点的部署任务失败时,会调用
  • v2_runner_on_unreachable:节点不可达时,会调用
  • v2_runner_on_skipped:部署任务跳过时,会调用
  • v2_playbook_on_stats:所有的部署任务完成时,调用
weixin

文章有帮助,请作者喝杯咖啡?

发表评论

邮箱地址不会被公开。 必填项已用*标注