Python調用ansible API系列（三）帶有callback的執行adhoc和playbook

在第二篇文章中雖然能夠執行adhoc和playbook可是執行結果的輸出並非特別直觀，雖然沒有報錯可是到底什麼結果其實你是不知道的尤爲是在執行adhoc的時候，這時候咱們要利用callback來設置一下執行結果的輸出。

執行adhoc

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from collections import namedtuple
# 核心類
# 用於讀取YAML和JSON格式的文件
import sys
from ansible.parsing.dataloader import DataLoader
# 用於存儲各種變量信息
from ansible.vars.manager import VariableManager
# 用於導入資產文件
from ansible.inventory.manager import InventoryManager
# 存儲執行hosts的角色信息
from ansible.playbook.play import Play
# ansible底層用到的任務隊列
from ansible.executor.task_queue_manager import TaskQueueManager
# 狀態回調，各類成功失敗的狀態
from ansible.plugins.callback import CallbackBase


class MyCallbackBase(CallbackBase):
    """
    經過api調用ac-hoc的時候輸出結果不少時候不是很明確或者說不是咱們想要的結果，主要它仍是輸出到STDOUT，並且一般咱們是在工程裏面執行
    這時候就須要後臺的結果前端能夠解析，正常的API調用輸出前端很難解析。 對比以前的執行 adhoc()查看區別。
    爲了實現這個目的就須要重寫CallbackBase類，須要重寫下面三個方法
    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)  # python3中重載父類構造方法的方式，在Python2中寫法會有區別。
        self.host_ok = {}
        self.host_unreachable = {}
        self.host_failed = {}

    def v2_runner_on_unreachable(self, result):
        """
        重寫 unreachable 狀態
        :param result:  這是父類裏面一個對象，這個對象能夠獲取執行任務信息
        """
        self.host_unreachable[result._host.get_name()] = result

    def v2_runner_on_ok(self, result, *args, **kwargs):
        """
        重寫 ok 狀態
        :param result:
        """
        self.host_ok[result._host.get_name()] = result

    def v2_runner_on_failed(self, result, *args, **kwargs):
        """
        重寫 failed 狀態
        :param result:
        """
        self.host_failed[result._host.get_name()] = result


def useMyCallbackBase():
    """
    這裏經過調用ad-hoc來使用自定義callback
    :return:
    """
    dl = DataLoader()
    im = InventoryManager(loader=dl, sources=["hosts"])
    vm = VariableManager(loader=dl, inventory=im)
    Options = namedtuple("Options", [
        "connection", "remote_user", "ask_sudo_pass", "verbosity", "ack_pass",
        "module_path", "forks", "become", "become_method", "become_user", "check",
        "listhosts", "listtasks", "listtags", "syntax", "sudo_user", "sudo", "diff"
    ])
    options = Options(connection='smart', remote_user=None, ack_pass=None, sudo_user=None, forks=5, sudo=None,
                      ask_sudo_pass=False,
                      verbosity=5, module_path=None, become=None, become_method=None, become_user=None, check=False,
                      diff=False,
                      listhosts=None, listtasks=None, listtags=None, syntax=None)
    play_source = dict(name="Ansible Play",  # 任務名稱
                       hosts="test",  # 目標主機，能夠填寫具體主機也能夠是主機組名稱
                       gather_facts="no",  # 是否收集配置信息

                       # tasks是具體執行的任務，列表形式，每一個具體任務都是一個字典
                       tasks=[
                           dict(action=dict(module="shell", args="touch /tmp/bbb.txt", warn=False))
                       ])
    play = Play().load(play_source, variable_manager=vm, loader=dl)

    passwords = dict()  # 這個能夠爲空，由於在hosts文件中

    mycallback = MyCallbackBase()  # 實例化自定義callback

    tqm = TaskQueueManager(
        inventory=im,
        variable_manager=vm,
        loader=dl,
        options=options,
        passwords=passwords,
        stdout_callback=mycallback  # 配置使用自定義callback
    )
    tqm.run(play)
    # print(mycallback.host_ok.items())  # 它會返回2個東西，一個主機一個是執行結果對象
    # 定義數據結構
    result_raw = {"success": {}, "failed": {}, "unreachable": {}}
    # 若是成功那麼  mycallback.host_ok.items() 才能夠遍歷，上面的任務確定能成功因此咱們就直接遍歷這個
    for host, result in mycallback.host_ok.items():
        result_raw["success"][host] = result._result

    for host, result in mycallback.host_failed.items():
        result_raw["failed"][host] = result._result

    for host, result in mycallback.host_unreachable.items():
        result_raw["unreachable"][host] = result._result

    print(result_raw)


def main():
    useMyCallbackBase()

if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()

View Code

執行playbook

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import sys
from collections import namedtuple
# 核心類
# 用於讀取YAML和JSON格式的文件
from ansible.parsing.dataloader import DataLoader
# 用於存儲各種變量信息
from ansible.vars.manager import VariableManager
# 用於導入資產文件
from ansible.inventory.manager import InventoryManager
from ansible.executor.playbook_executor import PlaybookExecutor
# 狀態回調，各類成功失敗的狀態
from ansible.plugins.callback import CallbackBase


class MyPlaybookCallbackBase(CallbackBase):
    """
    playbook的callback改寫，格式化輸出playbook執行結果
    """
    CALLBACK_VERSION = 2.0

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.task_ok = {}
        self.task_unreachable = {}
        self.task_failed = {}
        self.task_skipped = {}
        self.task_status = {}

    def v2_runner_on_unreachable(self, result):
        """
        重寫 unreachable 狀態
        :param result:  這是父類裏面一個對象，這個對象能夠獲取執行任務信息
        """
        self.task_unreachable[result._host.get_name()] = result

    def v2_runner_on_ok(self, result, *args, **kwargs):
        """
        重寫 ok 狀態
        :param result:
        """
        self.task_ok[result._host.get_name()] = result

    def v2_runner_on_failed(self, result, *args, **kwargs):
        """
        重寫 failed 狀態
        :param result:
        """
        self.task_failed[result._host.get_name()] = result

    def v2_runner_on_skipped(self, result):
        self.task_skipped[result._host.get_name()] = result

    def v2_playbook_on_stats(self, stats):
        hosts = sorted(stats.processed.keys())
        for h in hosts:
            t = stats.summarize(h)
            self.task_status[h] = {
                "ok": t["ok"],
                "changed": t["changed"],
                "unreachable": t["unreachable"],
                "skipped": t["skipped"],
                "failed": t["failed"]
            }


def usecallbackplaybook():
    """
    調用 playbook
    調用playboo大體和調用ad-hoc相同，只是真正調用的是使用PlaybookExecutor
    :return:
    """
    dl = DataLoader()
    im = InventoryManager(loader=dl, sources=["hosts"])
    vm = VariableManager(loader=dl, inventory=im)
    Options = namedtuple("Options", [
        "connection", "remote_user", "ask_sudo_pass", "verbosity", "ack_pass",
        "module_path", "forks", "become", "become_method", "become_user", "check",
        "listhosts", "listtasks", "listtags", "syntax", "sudo_user", "sudo", "diff"
    ])
    options = Options(connection='smart', remote_user=None, ack_pass=None, sudo_user=None, forks=5, sudo=None,
                      ask_sudo_pass=False,
                      verbosity=5, module_path=None, become=None, become_method=None, become_user=None, check=False,
                      diff=False,
                      listhosts=None, listtasks=None, listtags=None, syntax=None)
    passwords = dict()  # 這個能夠爲空，由於在hosts文件中
    #
    try:
        playbook = PlaybookExecutor(playbooks=["f1.yml"], inventory=im, variable_manager=vm, loader=dl, options=options, passwords=passwords)
        callback = MyPlaybookCallbackBase()
        playbook._tqm._stdout_callback = callback  # 配置callback
        playbook.run()

        # print(callback.task_ok.items())
        result_raw = {"ok": {}, "failed": {}, "unreachable": {}, "skipped": {}, "status": {}}
        for host, result in callback.task_ok.items():
            result_raw["ok"][host] = result._result

        for host, result in callback.task_failed.items():
            result_raw["failed"][host] = result._result

        for host, result in callback.task_unreachable.items():
            result_raw["unreachable"][host] = result._result

        for host, result in callback.task_skipped.items():
            result_raw["skipped"][host] = result._result

        for host, result in callback.task_status.items():
            result_raw["status"][host] = result._result

        print(result_raw)
    except Exception as err:
        print(err)


def main():
    usecallbackplaybook()


if __name__ == "__main__":
    try:
        main()
    finally:
        sys.exit()

View Code