Big Picture

ansible 的用途和使用场景就不再赘述了，日常使用中难免会遇到一些定制化需求，比如对接自动化操作平台需要将ansible api化，对接一些其他平台/网络设备/存储设备等需要自己定制模块进行自动化运维等，所以ansible的二次开发比较常见。

开发类别

ansible 开发主要分为4大类:

api的调用&&重封装

modules的开发

plugin的开发

源码的修改

api的调用&&重封装

ansible 一般使用都是基于命令行的方式进行执行的，但对于有些公司需要将ansible作为自动化运维平台的底层技术的时候，就需要通过api的方式进行调用, 主要是需要重新编写一下callback(ansible返回)以及动态inventory。

举个例子：

#!/usr/bin/env python3#-*- coding:utf-8 -*-# author:barney# ansible version: ansible==3.2.0, ansible-base==2.10.7import reimport requestsimport jsonimport datetimefrom oslo_config import cfgfrom oslo_log import logimport datetimefrom vmc.controller import ControllerBase# ansible 相关# ansible 不支持eventlet的monkeypatch 对 select的封装，所以需要关闭import shutil  from ansible.module_utils.common.collections import ImmutableDict  from ansible.parsing.dataloader import DataLoader  from ansible.vars.manager import VariableManager  from ansible.inventory.manager import InventoryManager  from ansible.playbook.play import Play  from ansible.executor.task_queue_manager import TaskQueueManager  from ansible.plugins.callback import CallbackBase  from ansible import context  import ansible.constants as Cimport yamlLOG = log.getLogger(__name__)CONF = cfg.CONF  class ResultCallback(CallbackBase):    """      基于playbook, 需要对callbackBase类的部分方法      """    def __init__(self, *args, **kwargs):        super().__init__(*args, **kwargs)        self.host_ok = {
   }        self.host_unreachable = {
   }        self.host_failed = {
   }        self.task_ok = {
   }    def v2_runner_on_unreachable(self, result):        self.host_unreachable[result._host.get_name() + "-" + datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')] = result    def v2_runner_on_ok(self, result, **kwargs):        self.host_ok[result._host.get_name() + "-" + datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')] = result    def v2_runner_on_failed(self, result, **kwargs):        self.host_failed[result._host.get_name() + "-" + datetime.datetime.now().strftime('%Y%m%d%H%M%S%f')] = resultclass MyAnsiable():      def __init__(self,          connection='smart',        remote_user='ops',        remote_password=None,        private_key_file=None,        sudo=None,        sudo_user='root',        ask_sudo_pass=False,          module_path=None,        become=True,        become_method='sudo',        become_user='root',        check=False,         diff=False,          listhosts=None,        listtasks=None,        listtags=None,          verbosity=5,          syntax=None,          start_at_task=None,        forks=10,        timeout=5,        tags=[],        skip_tags=[],        inventory=None):        """          初始化函数，定义的默认的选项值，          在初始化的时候可以传参，以便覆盖默认选项的值          """          context.CLIARGS = ImmutableDict(              connection=connection,              remote_user=CONF.ansible.remote_user,              private_key_file=private_key_file,              sudo=sudo,              sudo_user=sudo_user,              ask_sudo_pass=ask_sudo_pass,              module_path=module_path,              become=become,              become_method=become_method,              become_user=become_user,              verbosity=verbosity,              listhosts=listhosts,              listtasks=listtasks,              listtags=listtags,              syntax=syntax,              start_at_task=start_at_task,             forks=CONF.ansible.forknum,            tags=tags,            skip_tags=skip_tags,            timeout=CONF.ansible.timeout        )        self.inventory = inventory if inventory else "localhost,"        # 实例化数据解析器          self.loader = DataLoader()        # 实例化 资产配置对象          self.inv_obj = InventoryManager(loader=self.loader, sources=self.inventory)        # 设置密码          self.passwords = remote_password        # 实例化回调插件对象          self.results_callback = ResultCallback()                # 变量管理器          self.variable_manager = VariableManager(self.loader, self.inv_obj)    def run(self, gether_facts="no", module="ping", args='', task_time=30):          """          参数说明：          task_time -- 执行异步任务时等待的秒数，这个需要大于 0 ，等于 0 的时候不支持异步（默认值）。这个值应该等于执行任务实际耗时时间为好        服务器数量少的情况下，使用同步即可         """          play_source =  dict(              name = "Ansible Used for Auto Mgr",              hosts = self.inventory,              gather_facts = gether_facts,              tasks = [                # 异步调用模板               # {"action":{"module": module, "args": args}, "async": task_time, "poll": 0}])               # 默认还是走同步调用               {
   "action":{
   "module": module, "args": args}}])        play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)        tqm = None        fin_result = []        try:              tqm = TaskQueueManager(                        inventory=self.inv_obj ,                        variable_manager=self.variable_manager,                        loader=self.loader,                        passwords=self.passwords,                        stdout_callback=self.results_callback)            result = tqm.run(play)                          for host,result in self.results_callback.host_ok.items():                tmp_result = {
   "host": host.split("-")[0], "status": "success","stdout": result._result["stdout"]}                fin_result.append(tmp_result)            for host,result in self.results_callback.host_failed.items():                tmp_result = {
   "host": host.split("-")[0], "status": "failed","stdout": result._result["stderr"]}                fin_result.append(tmp_result)            for host,result in self.results_callback.host_unreachable.items():                tmp_result = {
   "host": host.split("-")[0], "status": "unreachable","stdout": "connect timeout"}                fin_result.append(tmp_result)        finally:              if tqm is not None:                  tqm.cleanup()              shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)                return(fin_result)                def update_yml_host(self, ymlfile, hosts):        with open(ymlfile,'r',encoding='utf-8') as f:            result = f.read()            yml = yaml.load(result,Loader=yaml.FullLoader)            for section in yml:                section["hosts"] = hosts            with open(ymlfile,'w',encoding='utf-8') as w_f:                yaml.dump(yml,w_f)    def playbook(self, playbooks):          """        参数说明：        playbooks, 需要是一个列表类型          """        from ansible.executor.playbook_executor import PlaybookExecutor                playbookfile = playbooks[0]                 # 更新yml文件        self.update_yml_host(playbookfile, self.inventory)        playbook = PlaybookExecutor(playbooks=playbooks,                        inventory=self.inv_obj,                        variable_manager=self.variable_manager,                        loader=self.loader,                        passwords=self.passwords)        # 使用回调函数        playbook._tqm._stdout_callback = self.results_callback        result = playbook.run()                fin_result = []                """        todo:        增加tag标识，结果的每一步的tag 需要增加        """        # 判断成功的item,如果成功则并入对应host的步骤        # 成功情况处理               for host,result in self.results_callback.host_ok.items():            hn = host.split("-")[0]            exectime = host.split("-")[1]            if result._result.__contains__('ansible_facts'):                continue            if result._result.__contains__("stdout") and len(result._result["stdout"]) == 0:                stdout = "ok"            elif result._result.__contains__("stdout") and len(result._result["stdout"]) != 0:                stdout = result._result["stdout"]            else:                stdout = "ok"            tmp_result = {
   "exectime": exectime, "status": "success","stdout": stdout}            if self.judge_range(fin_result, hn):                for host_result in fin_result:                    if host_result['host'] == hn:                        host_result['step'].append(tmp_result)                        break            else:                fin_result.append({
   "host": hn, "status":"success", "step": [tmp_result]})                   # 失败情况处理        for host,result in self.results_callback.host_failed.items():            hn = host.split("-")[0]            exectime = host.split("-")[1]            if result._result.__contains__('ansible_facts'):                continue            if result._result.__contains__("stderr"):                stdout = result._result["stderr"]            else:                stdout = "unknown error"            tmp_result = {
   "exectime": exectime, "status": "failed","stdout": stdout}            if self.judge_range(fin_result, hn):                for host_result in fin_result:                    if host_result['host'] == hn:                        host_result["status"] = "failed"                        host_result['step'].append(tmp_result)                        break            else:                fin_result.append({
   "host": hn, "status":"failed", "step": [tmp_result]})        # 超时情况处理        for host,result in self.results_callback.host_unreachable.items():            hn = host.split("-")[0]            exectime = host.split("-")[1]            if result._result.__contains__('ansible_facts'):                continue            stdout = "connect timeout"            tmp_result = {
   "exectime": exectime, "status": "unreachable","stdout": stdout}            if self.judge_range(fin_result, hn):                for host_result in fin_result:                    if host_result['host'] == hn:                        host_result['step'].append(tmp_result)                        break            else:                fin_result.append({
   "host": hn, "status":"unreachable", "step": [tmp_result]})        return fin_result    def judge_range(self, fin, hn):        for x in fin:            if x["host"] == hn:                return True        return False

个人公众号, 分享一些日常开发，运维工作中的日常以及一些学习感悟，欢迎大家互相学习，交流

在这里插入图片描述

转载地址：https://blog.csdn.net/zyxpaomian/article/details/116530677 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！

上一篇：ansible开发(2)

下一篇：K8S水平伸缩-HPA

发表评论

关于作者

喝酒易醉，品茶养心，人生如梦，品茶悟道，何以解忧？唯有杜康！

-- 愿君每日到此一游！