Ansible python API使用实例

Ansible 是一个开源的自动化运维工具，可以实现批量主机命令、脚本等的批量执行，方便快捷，被管理主机不需要部署代理，具体请见官网
https://www.ansible.com/ <https://www.ansible.com/>

这里简单介绍一下其提供的python
API如何使用。Ansible目前已经发布多个版本，不同版本的API变化较大，在2.0的版本以前，ansible的API十分简单。通过大概十几行代码就可以模拟通过ad-hoc的方式运行annsible命令的。但是在2.0及以后的版本，难度突然陡增，与此同时更多的更加底层的东西被开放给一般开发，虽然复杂了，但是也更加灵活了。

2.0以前使用请参考https://www.cnblogs.com/franknihao/p/9009110.html
<https://www.cnblogs.com/franknihao/p/9009110.html>

本文介绍的API是基于ansible 2.6以上。
#!/usr/bin/env python from collections import namedtuple from
ansible.executor.task_queue_manager import TaskQueueManager from
ansible.inventory import Inventory from ansible.parsing.dataloader import
DataLoader from ansible.playbook.play import Play from ansible.plugins.callback
import CallbackBase from ansible.vars import VariableManager from
ansible.executor.playbook_executor import PlaybookExecutor # Create a callback
object so we can capture the output class ResultsCollector(CallbackBase): def
__init__(self, *args, **kwargs): super(ResultsCollector, self).__init__(*args,
**kwargs) self.host_ok = {} self.host_unreachable = {} self.host_failed = {}
def v2_runner_on_unreachable(self, result):
self.host_unreachable[result._host.get_name()] = result def
v2_runner_on_ok(self, result, *args, **kwargs):
self.host_ok[result._host.get_name()] = result def v2_runner_on_failed(self,
result, *args, **kwargs): self.host_failed[result._host.get_name()] = result
def order_run(ip, p_name, p_module, p_args): host_list = [ip] Options =
namedtuple('Options', ['connection', 'module_path', 'forks', 'remote_user',
'private_key_file', 'ssh_common_args', 'ssh_extra_args', 'sftp_extra_args',
'scp_extra_args', 'become', 'become_method', 'become_user', 'verbosity',
'check', 'diff']) # required for #
https://github.com/ansible/ansible/blob/devel/lib/ansible/inventory/manager.py#L204
sources = ','.join(host_list) if len(host_list) == 1: sources += ',' #
initialize needed objects loader = DataLoader() options =
Options(connection='smart', module_path=None, forks=1, remote_user=None,
private_key_file=None, ssh_common_args=None, ssh_extra_args=None,
sftp_extra_args=None, scp_extra_args=None, become=None, become_method=None,
become_user=None, verbosity=None, check=False, diff=False) passwords = dict() #
create inventory and pass to var manager variable_manager = VariableManager()
inventory = Inventory(loader=loader, variable_manager=variable_manager,
host_list=sources) variable_manager.set_inventory(inventory) # create play with
tasks play_source = dict( name=p_name, hosts=host_list, gather_facts='no',
tasks=[dict(action=dict(module=p_module, args=p_args))] ) play =
Play().load(play_source, variable_manager=variable_manager, loader=loader) #
actually run it tqm = None callback = ResultsCollector() try: tqm =
TaskQueueManager( inventory=inventory, variable_manager=variable_manager,
loader=loader, options=options, passwords=passwords, ) tqm._stdout_callback =
callback result = tqm.run(play) finally: if tqm is not None: tqm.cleanup() '''
print("UP ***********") for host, result in callback.host_ok.items(): print
result._result['stdout'] print("FAILED *******") for host, result in
callback.host_failed.items(): print result._result['msg'] print("DOWN
*********") for host, result in callback.host_unreachable.items(): print
result._result['msg'] ''' up=faild=down=None for host, result in
callback.host_ok.items(): up=result._result['stdout'] ''' for host, result in
callback.host_failed.items(): faild=result._result['msg'] for host, result in
callback.host_unreachable.items(): down=result._result['msg'] ''' res={'UP':up,
'FAILD':faild, 'DOWN':down} return res def playbook_run(host, playbook_path):
host_list = [host] sources = ','.join(host_list) if len(host_list) == 1:
sources += ',' loader = DataLoader() variable_manager = VariableManager()
inventory = Inventory(loader=loader, variable_manager=variable_manager,
host_list=sources) variable_manager.set_inventory(inventory) Options =
namedtuple('Options', ['listtags', 'listtasks', 'listhosts', 'syntax',
'connection','module_path', 'forks', 'remote_user',
'private_key_file','ssh_common_args','ssh_extra_args', 'sftp_extra_args',
'scp_extra_args', 'become', 'become_method', 'become_user', 'verbosity',
'check','diff']) options = Options(listtags=False, listtasks=False,
listhosts=False, syntax=False, connection='smart', module_path=None, forks=1,
remote_user=None, private_key_file=None, ssh_common_args =None,
ssh_extra_args=None, sftp_extra_args=None, scp_extra_args=None, become=True,
become_method=None, become_user=None, verbosity=None, check=False, diff=False)
passwords = {} callback = ResultsCollector() pbex =
PlaybookExecutor(playbooks=[playbook_path], inventory=inventory,
variable_manager=variable_manager, loader=loader, options=options,
passwords=passwords) pbex._tqm._stdout_callback = callback result = pbex.run()
''' print("UP ***********") for host, result in callback.host_ok.items(): print
result._result['stdout'] print("FAILED *******") for host, result in
callback.host_failed.items(): print result._result['msg'] print("DOWN
*********") for host, result in callback.host_unreachable.items(): print
result._result['msg'] ''' up=faild=down=None for host, result in
callback.host_ok.items(): up=result._result['stdout'] ''' for host, result in
callback.host_failed.items(): faild=result._result['msg'] for host, result in
callback.host_unreachable.items(): down=result._result['msg'] ''' res={'UP':up,
'FAILD':faild, 'DOWN':down} return res #调用 res=control.order_run(hostip, 'get
vmcore log', 'shell', 'cat /var/crash/*/vmcore-dmesg.txt')
res=control.playbook_run(hostip, yml_path)
 

• « 上一篇：10分钟教你理解反射
• » 下一篇：每周实习总结之HTML和CSS实战一