在python中使用Ansible实现Devops的相关总结

前言

Ansible虽然底层是用python写的，但是对python的API支持并不友好，官方文档里只用了一个example来说明，想知道各个组件的参数功能可能只能去看源码了，现在记录一下自己的总结

环境——ansible==2.8.1

问题总结

如果需要有额外的参数，或者在playbook中需要传入参数怎么办？

在context.CLIARGS = ImmutableDict中通过extra_vars=[extra_vars]的方式
ansible.cfg文件应该放在哪里？

ansible.cfg这个文件pip install后并不会自动在/etc/ansible中创建，需要去github中将里面的example文件夹下的文件拷贝进本地才行。优先级问题：这个文件如果需要跟着项目走，那么在哪里执行run脚本文件，这个ansible.cfg文件就应该放在哪里。如果在这里找不到就去/etc/ansible里找，当然也可以设置环境变量，这个优先级最高
如果playbook一个命令执行时间太长怎么办？

ssh链接最长好像只有5分钟，5分钟后还没执行完，直接返回unreachable。我尝试在代码中添加constant，或者修改ansible.cfg里的各种timeout，或者retries都不管用，最后是通过在playbook中加入async解决的

例子：
```
---
- name: 666hosts: "666"gather_facts: notasks:- name: 6不6shell:chdir: /rootcmd: lsasync: 1200poll: 10
```
- poll的意思是说每隔10秒重新连接一次，只要poll大于0，就是阻塞状态，直到这个异步任务执行结束才会执行下一个task。如果设置poll=0，就会即可跳过这个task（转到后台执行），继续执行下面的任务，使用于两种不同的场景，详情见官方手册
参考：

https://stackoverflow.com/questions/41455002/long-running-command-in-ansible-ending-in-failed-status-with-host-unreachable

https://www.axelerant.com/resources/team-blog/how-handle-long-running-tasks-ansible
关于callback函数的改写

callback函数是说每次有一个task任务被执行，会根据执行的状态success，failed等等去调用这些函数，result就是执行中调用这个函数时的入参，通过下面的例子的方式可以保存下来这些内容

实例：

import json
import shutil
from ansible import constants
from ansible.executor.playbook_executor import PlaybookExecutor
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.inventory.group import Group
from ansible.inventory.host import Host
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase
from ansible import context
import ansible.constants as Cclass ResultCallback(CallbackBase):"""重写callbackBase类的部分方法"""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_ok(self, result, **kwargs):self.task_ok.append({result._host.get_name(): result})def v2_runner_on_failed(self, result, **kwargs):self.task_failed.append({result._host.get_name(): result})def v2_runner_on_skipped(self, result, **kwargs):self.task_skipped.append({result._host.get_name(): result})def v2_runner_on_unreachable(self, result, **kwargs):self.task_unreachable.append({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"]}class MyInventory:def __init__(self, hostsresource):"""初始化函数:param hostsresource: 主机资源可以有2种形式列表形式: [{"ip": "172.16.48.171", "port": "22", "username": "root", "password": "123456"}]字典形式: {"Group1": {"hosts": [{"ip": "192.168.200.10", "port": "1314", "username": "root", "password": None}],"vars": {"var1": "ansible"}},"Group2": {}}"""self._hostsresource = hostsresourceself._loader = DataLoader()self._hostsfilelist = ["temphosts"]"""sources这个我们知道这里是设置hosts文件的地方，它可以是一个列表里面包含多个文件路径且文件真实存在，在单纯的执行ad-hoc的时候这里的文件里面必须具有有效的hosts配置，但是当通过动态生成的资产信息的时候这个文件必须存在但是它里面可以是空的，如果这里配置成None那么它不影响资产信息动态生成但是会有一个警告，所以还是要配置一个真实文件。"""self._inventory = InventoryManager(loader=self._loader, sources=self._hostsfilelist)self._variable_manager = VariableManager(loader=self._loader, inventory=self._inventory)self._dynamic_inventory()def _add_dynamic_group(self, hosts_list, groupname, groupvars=None):"""动态添加主机到指定的主机组完整的HOSTS文件格式[test1]hostname ansible_ssh_host=192.168.1.111 ansible_ssh_user="root" ansible_ssh_pass="123456"但通常我们都省略hostname，端口也省略因为默认是22，这个在ansible配置文件中有，除非有非22端口的才会配置[test1]192.168.100.10 ansible_ssh_user="root" ansible_ssh_pass="123456" ansible_python_interpreter="/PATH/python3/bin/python3":param hosts_list: 主机列表 [{"ip": "192.168.100.10", "port": "22", "username": "root", "password": None}, {}]:param groupname:  组名称:param groupvars:  组变量，格式为字典:return:"""# 添加组my_group = Group(name=groupname)self._inventory.add_group(groupname)# 添加组变量if groupvars:for key, value in groupvars.items():my_group.set_variable(key, value)# 添加一个主机for host in hosts_list:hostname = host.get("hostname", None)hostip = host.get("ip", None)if hostip is None:print("IP地址为空，跳过该元素。")continuehostport = host.get("port", "22")username = host.get("username", "root")password = host.get("password", None)ssh_key = host.get("ssh_key", None)python_interpreter = host.get("python_interpreter", None)try:# hostname可以不写，如果为空默认就是IP地址if hostname is None:hostname = hostip# 生成一个host对象my_host = Host(name=hostname, port=hostport)# 添加主机变量self._variable_manager.set_host_variable(host=my_host, varname="ansible_ssh_host", value=hostip)self._variable_manager.set_host_variable(host=my_host, varname="ansible_ssh_port", value=hostport)if password:self._variable_manager.set_host_variable(host=my_host, varname="ansible_ssh_pass", value=password)self._variable_manager.set_host_variable(host=my_host, varname="ansible_ssh_user", value=username)if ssh_key:self._variable_manager.set_host_variable(host=my_host, varname="ansible_ssh_private_key_file", value=ssh_key)if python_interpreter:self._variable_manager.set_host_variable(host=my_host, varname="ansible_python_interpreter", value=python_interpreter)# 添加其他变量for key, value in host.items():if key not in ["ip", "hostname", "port", "username", "password", "ssh_key", "python_interpreter"]:self._variable_manager.set_host_variable(host=my_host, varname=key, value=value)# 添加主机到组self._inventory.add_host(host=hostname, group=groupname, port=hostport)except Exception as err:print(err)def _dynamic_inventory(self):"""添加 hosts 到inventory:return:"""if isinstance(self._hostsresource, list):self._add_dynamic_group(self._hostsresource, "default_group")elif isinstance(self._hostsresource, dict):for groupname, hosts_and_vars in self._hostsresource.items():self._add_dynamic_group(hosts_and_vars.get("hosts"), groupname, hosts_and_vars.get("vars"))@propertydef INVENTORY(self):"""返回资产实例:return:"""return self._inventory@propertydef VARIABLE_MANAGER(self):"""返回变量管理器实例:return:"""return self._variable_managerclass MyAnsiable():def __init__(self,hostsresource=None,connection='smart',  # 连接方式 local 本地方式，smart ssh方式remote_user=None,  # ssh 用户remote_password=None,  # ssh 用户的密码，应该是一个字典, key 必须是 conn_passprivate_key_file=None,  # 指定自定义的私钥地址sudo=None, sudo_user=None, ask_sudo_pass=None,module_path=None,  # 模块路径，可以指定一个自定义模块的路径become=None,  # 是否提权become_method=None,  # 提权方式 默认 sudo 可以是 subecome_user=None,  # 提权后，要成为的用户，并非登录用户check=False, diff=False,listhosts=None,listtasks=None,listtags=None,verbosity=3,forks=5,syntax=None,start_at_task=None,inventory=None,extra_vars=None):# 函数文档注释"""初始化函数，定义的默认的选项值，在初始化的时候可以传参，以便覆盖默认选项的值"""context.CLIARGS = ImmutableDict(connection=connection,remote_user=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,forks=forks,become_method=become_method,become_user=become_user,verbosity=verbosity,listhosts=listhosts,listtasks=listtasks,listtags=listtags,syntax=syntax,start_at_task=start_at_task,check=False,extra_vars=[extra_vars]  # 如果有附加参数，或者playbook中需要传入参数，可以使用这个功能)# 三元表达式，假如没有传递 inventory, 就使用 "localhost,"# 指定 inventory 文件# inventory 的值可以是一个 资产清单文件# 也可以是一个包含主机的元组，这个仅仅适用于测试#  比如 ： 1.1.1.1,    # 如果只有一个 IP 最后必须有英文的逗号#  或者： 1.1.1.1, 2.2.2.2my_inventory = MyInventory(hostsresource=hostsresource)# 实例化数据解析器self.loader = DataLoader()# 实例化 资产配置对象self.inv_obj = my_inventory.INVENTORY# 设置密码self.passwords = remote_password# 实例化回调插件对象self.results_callback = ResultCallback()# 变量管理器self.variable_manager = my_inventory.VARIABLE_MANAGERdef run(self, hosts='localhost', gether_facts="no", module="ping", args='', task_time=0):"""参数说明：task_time -- 执行异步任务时等待的秒数，这个需要大于 0 ，等于 0 的时候不支持异步（默认值）。这个值应该等于执行任务实际耗时时间为好"""play_source = dict(name="Ad-hoc",hosts=hosts,gather_facts=gether_facts,tasks=[# 这里每个 task 就是这个列表中的一个元素，格式是嵌套的字典# 也可以作为参数传递过来，这里就简单化了。{"action": {"module": module, "args": args}, "async": task_time, "poll": 0}])play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)tqm = Nonetry: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)finally:if tqm is not None:tqm.cleanup()shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)def playbook(self, playbooks):"""Keyword arguments:playbooks --  需要是一个列表类型"""try: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# 下面这些设置只有constants.HOST_KEY_CHECKING = False管用# constants.HOST_KEY_CHECKING = False# constants.CONNECT_TIMEOUT = 1200# constants.COMMAND_TIMEOUT = 1200# constants.ANSIBLE_PERSISTENT_CONNECT_TIMEOUT = 1200# constants.ANSIBLE_PERSISTENT_COMMAND_TIMEOUT = 1200# constants.TIMEOUT = 1200# constants.CONFIG_FILE = './888/ansible.cfg'result = playbook.run()except Exception:raisedef get_result(self):result_raw = dict()for i in self.results_callback.task_ok:for host, result in i.items():if result_raw.get(host):result_raw[host]["success"].update({result._task.get_name(): result._result})else:result_raw[host] = dict()result_raw[host]["success"] = {result._task.get_name(): result._result}for i in self.results_callback.task_failed:for host, result in i.items():if result_raw.get(host):result_raw[host]["failed"].update({result._task.get_name(): result._result})else:result_raw[host] = dict()result_raw[host]["failed"] = {result._task.get_name(): result._result}for i in self.results_callback.task_unreachable:for host, result in i.items():if result_raw.get(host):result_raw[host]["unreachable"].update({result._task.get_name(): result._result})else:result_raw[host] = dict()result_raw[host]["unreachable"] = {result._task.get_name(): result._result}for i in self.results_callback.task_skipped:for host, result in i.items():if result_raw.get(host):result_raw[host]["skipped"].update({result._task.get_name(): result._result})else:result_raw[host] = dict()result_raw[host]["skipped"] = {result._task.get_name(): result._result}for host, result in self.results_callback.task_status.items():result_raw[host]["status"] = result# 最终打印结果，并且使用 JSON 继续格式化print(json.dumps(result_raw, indent=4))return json.dumps(result_raw)

callback输出：

{"666.666.666.666":{"success":{"task_name":result_result(这是一个字典)}}
}

在python中使用Ansible实现Devops的相关总结相关推荐

Python中数字以及算数运算符的相关使用
Python数字数字数据类型用于存储数值. 他们是不可改变的数据类型,这意味着改变数字数据类型会分配一个新的对象. 当你指定一个值时,Number对象就会被创建: <span style=&q ...
set在python中什么意思_python中set是什么意思
python中set是什么意思,对象,还可以,技术文章,创建一个,使用方法 python中set是什么意思易采站长站,站长之家为您整理了python中set是什么意思的相关内容. set()函数创建 ...
功能式Python中的探索性数据分析
欢迎大家前往腾讯云+社区,获取更多腾讯海量技术实践干货哦~ 这里有一些技巧来处理日志文件提取.假设我们正在查看一些Enterprise Splunk提取.我们可以用Splunk来探索数据.或者我们可以 ...
python中什么是主要数据类型_python中基本数据类型是什么
python中基本数据类型是什么,变量,整型,类型,赋值,等号 python中基本数据类型是什么易采站长站,站长之家为您整理了python中基本数据类型是什么的相关内容. python中基本数据类型 ...
python 中的metaclass和baseclasses
提前说明: class object 指VM中的class 对象,因为python一切对象,class在VM也是一个对象,需要区分class对象和 class实例对象. class instance ...
python变量的理解_如何理解Python中的变量
在本篇文章里小编给大家分享的是关于Python中变量是什么意思的相关基础知识点,需要的朋友们可以学习下. 变量在Python中,存储一个数据,需要定义一个变量 number1 = 1 #numbe1 ...
Python中sort和sorted函数代码解析
Python中sort和sorted函数代码解析本文研究的主要是Python中sort和sorted函数的相关内容,具体如下. 一.sort函数 sort函数是序列的内部函数函数原型: L.sor ...
python中int函数是什么作用_python中int函数怎么用
python中int函数怎么用,字符串,函数,数字,出现在,赋值 python中int函数怎么用易采站长站,站长之家为您整理了python中int函数怎么用的相关内容. int() 函数用于将一个字 ...
python中re.compile_什么是pythonre.compile函数？
在这篇文章之中我们来了解一下关于python re.compile的知识,有些朋友可能是刚刚接触到python这一编程语言,对这一方面不是特别的了解,但是没关系接下来这篇文章将会来带大家来了解关于py ...

在python中使用Ansible实现Devops的相关总结

在python中使用Ansible实现Devops的相关总结

前言

环境——ansible==2.8.1

问题总结

实例：

相关参考资料

在python中使用Ansible实现Devops的相关总结相关推荐

最新文章

热门文章