05.Ansible Python API
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了05.Ansible Python API，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含16009字，纯文字阅读大概需要23分钟。
内容图文
Ansible Python API
- 官方示例
#!/usr/bin/env python
import json
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 C
class ResultCallback(CallbackBase):
"""A sample callback plugin used for performing an action as results come in
If you want to collect all results into a single object for processing at
the end of the execution, look into utilizing the ``json`` callback plugin
or writing your own custom callback plugin
"""
def v2_runner_on_ok(self, result, **kwargs):
"""Print a json representation of the result
This method could store the result in an instance attribute for retrieval later
"""
host = result._host
print(json.dumps({host.name: result._result}, indent=4))
# since the API is constructed for CLI it expects certain options to always be set in the context object
context.CLIARGS = ImmutableDict(connection='local', module_path=['/to/mymodules'], forks=10, become=None,
become_method=None, become_user=None, check=False, diff=False)
# initialize needed objects
loader = DataLoader() # Takes care of finding and reading yaml, json and ini files
passwords = dict(vault_pass='secret')
# Instantiate our ResultCallback for handling results as they come in. Ansible expects this to be one of its main display outlets
results_callback = ResultCallback()
# create inventory, use path to host config file as source or hosts in a comma separated string
inventory = InventoryManager(loader=loader, sources='localhost,')
# variable manager takes care of merging all the different sources to give you a unified view of variables available in each context
variable_manager = VariableManager(loader=loader, inventory=inventory)
# create data structure that represents our play, including tasks, this is basically what our YAML loader does internally.
play_source = dict(
name = "Ansible Play",
hosts = 'localhost',
gather_facts = 'no',
tasks = [
dict(action=dict(module='shell', args='ls'), register='shell_out'),
dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
]
)
# Create play object, playbook objects use .load instead of init or new methods,
# this will also automatically create the task objects from the info provided in play_source
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# Run it - instantiate task queue manager, which takes care of forking and setting up all objects to iterate over host list and tasks
tqm = None
try:
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=variable_manager,
loader=loader,
passwords=passwords,
stdout_callback=results_callback, # Use our custom callback instead of the ``default`` callback plugin, which prints to stdout
)
result = tqm.run(play) # most interesting data for a play is actually sent to the callback's methods
finally:
# we always need to cleanup child procs and the structures we use to communicate with them
if tqm is not None:
tqm.cleanup()
# Remove ansible tmpdir
shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)
- 执行ad-hoc任务
#!/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.inventory.host import Host
# 操作单个主机组信息
from ansible.inventory.group import Group
# 存储执行hosts的角色信息
from ansible.playbook.play import Play
# ansible底层用到的任务队列
from ansible.executor.task_queue_manager import TaskQueueManager
# 核心类执行playbook
from ansible.executor.playbook_executor import PlaybookExecutor
def adhoc():
"""
ad-hoc 调用
资产配置信息 这个是通过 InventoryManager和VariableManager 定义
执行选项 这个是通过namedtuple来定义
执行对象和模块 通过dict()来定义
定义play 通过Play来定义
最后通过 TaskQueueManager 的实例来执行play
:return:
"""
# 资产配置信息
dl = DataLoader()
im = InventoryManager(loader=dl, sources=["hosts"])
vm = VariableManager(loader=dl, inventory=im)
# 执行选项，这个类不是ansible的类，这个的功能就是为了构造参数
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的实例，然后你就可以赋值，这个为了给ansible设置执行选项 ansibile 172.16.48.171 -m shell -a 'ls /tmp' -f 5
这里的选项就是ansible命令中 -f -C -D -m等执行选项
"""
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的执行对象和模块，这里设置hosts，其实是因为play把play_source和资产信息关联后，执行的play的时候它会去资产信息中设置的sources的hosts文件中
# 找你在play_source中设置的hosts是否在资产管理类里面。
play_source = dict(name="Ansible Play", # 任务名称
hosts="172.16.48.242", # 目标主机，可以填写具体主机也可以是主机组名称
gather_facts="no", # 是否收集配置信息
# tasks是具体执行的任务，列表形式，每个具体任务都是一个字典
tasks=[
dict(action=dict(module="shell", args="ls /tmp"))
])
# 定义play
play = Play().load(play_source, variable_manager=vm, loader=dl)
passwords = dict() # 这个可以为空，因为在hosts文件中
#
tqm = TaskQueueManager(
inventory=im,
variable_manager=vm,
loader=dl,
options=options,
passwords=passwords,
)
result = tqm.run(play)
print(result)
def main():
adhoc()
if __name__ == "__main__":
try:
main()
finally:
sys.exit()
- 执行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.inventory.host import Host
# 操作单个主机组信息
from ansible.inventory.group import Group
# 存储执行hosts的角色信息
from ansible.playbook.play import Play
# ansible底层用到的任务队列
from ansible.executor.task_queue_manager import TaskQueueManager
# 核心类执行playbook
from ansible.executor.playbook_executor import PlaybookExecutor
def execplaybook():
"""
调用 playbook
调用playboo大致和调用ad-hoc相同，只是真正调用的是使用PlaybookExecutor
:return:
"""
# 资产配置信息
dl = DataLoader()
im = InventoryManager(loader=dl, sources=["hosts"])
vm = VariableManager(loader=dl, inventory=im)
# 执行选项，这个类不是ansible的类，这个的功能就是为了构造参数
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的实例，然后你就可以赋值，这个为了给ansible设置执行选项 ansibile 172.16.48.171 -m shell -a 'ls /tmp' -f 5
这里的选项就是ansible命令中 -f -C -D -m等执行选项
"""
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:
# playbooks参数里面放的就是playbook文件路径
playbook = PlaybookExecutor(playbooks=["f1.yml"], inventory=im, variable_manager=vm, loader=dl, options=options, passwords=passwords)
playbook.run()
except Exception as err:
print(err)
def main():
execplaybook()
if __name__ == "__main__":
try:
main()
finally:
sys.exit()
- ansible管理iptables
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""ansible管理iptables"""
import shutil
import ansible.constants as C
from ansible import context
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.inventory.manager import InventoryManager
from ansible.module_utils.common.collections import ImmutableDict
from ansible.parsing.dataloader import DataLoader
from ansible.playbook.play import Play
from ansible.plugins.callback import CallbackBase
from ansible.vars.manager import VariableManager
class ResultCallback(CallbackBase):
"""
回调类，给ansible执行ad-hoc命令后回调使用
"""
def __init__(self):
super(ResultCallback, self).__init__()
self.result = None
def v2_runner_on_ok(self, result, **kwargs):
"""
运行成功时回调
:param result:
:param kwargs:
:return: 返回json对象
"""
host = result._host
self.result = {host.name: result._result}
def v2_runner_on_failed(self, result, ignore_errors=False):
"""
运行失败时回调
:param result:
:param ignore_errors:
:return: 返回json对象
"""
host = result._host
self.result = {host.name: result._result}
def v2_runner_on_unreachable(self, result):
"""
主机连接失败时回调
:param result:
:return: 返回json对象
"""
host = result._host
self.result = {host.name: result._result}
class FireWall(object):
"""
iptables规则核心类
"""
def __init__(self, inventory_path, hosts='all', remote_user=None, become_user='root', password=None):
"""
:param inventory_path: str inventory文件路径 必传参数
:param hosts: str 传入的主机，和inventory文件的需要匹配上 默认所有主机
:param remote_user: str 远程连接用户 非必传参数 默认app用户
:param become_user: str 远程执行命令用户 非必传参数 默认root用户
:param password: str 远程主机密码 非必传参数 默认使用ssh公钥对连接
"""
context.CLIARGS = ImmutableDict(
connection='smart',
remote_user=remote_user,
ack_pass=None,
sudo_user=None,
forks=5,
sudo=True,
ask_sudo_pass=False,
verbosity=5,
module_path=None,
become=True,
become_method='sudo',
become_user=become_user,
check=False,
diff=False,
listhosts=None,
listtasks=None,
listtags=None,
syntax=None
)
self._loader = DataLoader()
self._passwords = dict() if password is None else dict(vault_pass=password)
self._results_callback = ResultCallback()
self._inventory = InventoryManager(loader=self._loader, sources=[inventory_path])
self._variable_manager = VariableManager(loader=self._loader, inventory=self._inventory)
self._hosts = hosts
self._redis_key = 'bdc_aomp_firewall_iptables_{}'.format('_'.join(self._hosts.split('.')))
def _run_shell_task(self, name, cmd):
"""
执行shell命令私有类
:param name: 任务名称
:param cmd: shell命令
:return: 返回回调类的结果
"""
play_source = dict(
name=name,
hosts=self._hosts,
gather_facts='no',
tasks=[
dict(action=dict(module='shell', args=cmd), register='shell_out'),
]
)
play = Play().load(play_source, variable_manager=self._variable_manager, loader=self._loader)
tqm = None
try:
tqm = TaskQueueManager(
inventory=self._inventory,
variable_manager=self._variable_manager,
loader=self._loader,
passwords=self._passwords,
stdout_callback=self._results_callback,
)
tqm.run(play)
finally:
if tqm is not None:
tqm.cleanup()
shutil.rmtree(C.DEFAULT_LOCAL_TMP, True)
return self._results_callback.result
def _save_rule(self):
"""
用来保存iptables的规则方法，默认添加和删除的iptables规则保存在内存中，执行完新增和删除规则后需要调用该方法
:return: tuple (True, None)
"""
cmd = 'service iptables save'
res = self._run_shell_task('append rules', cmd)
if res[self._hosts]['rc'] != 0:
return False, res[self._hosts]['stderr']
else:
return True, None
def get_iptables(self):
"""
通过ansible获取远程主机的iptables规则
:return: tuple (rules, None)
"""
name = 'get iptables'
cmd = 'iptables -nL --line-number'
result = self._run_shell_task(name, cmd)
if result is None:
return None, 'get iptables failed!'
if result[self._hosts].get('unreachable') is True:
return None, result[self._hosts]['msg']
if result[self._hosts]['rc'] != 0:
return None, result[self._hosts]['stderr']
try:
forward_index = result[self._hosts]['stdout_lines'].index('Chain FORWARD (policy DROP)')
except Exception:
forward_index = result[self._hosts]['stdout_lines'].index('Chain FORWARD (policy ACCEPT)')
output_index = result[self._hosts]['stdout_lines'].index('Chain OUTPUT (policy ACCEPT)')
input_list = []
forward_list = []
output_list = []
for i, data in enumerate(result[self._hosts]['stdout_lines']):
res_dict = {
'num': None,
'target': None,
'prot': None,
'opt': None,
'source': None,
'destination': None,
'rules': None
}
if len(data) > 0 and 'num' not in data and 'Chain' not in data:
rules = data.split()
res_dict['num'] = rules[0]
res_dict['target'] = rules[1]
res_dict['prot'] = rules[2]
res_dict['opt'] = rules[3]
res_dict['source'] = rules[4]
res_dict['destination'] = rules[5]
if len(rules) > 6:
res_dict['rules'] = ' '.join(rules[6:])
if i < forward_index:
input_list.append(res_dict)
elif i < output_index:
forward_list.append(res_dict)
else:
output_list.append(res_dict)
res = {"input": input_list, "forward": forward_list, "output": output_list}
return res, None
def get_hosts(self):
"""
获取管理的主机列表
:return: dict
"""
return {'hosts': [i.address for i in self._inventory.get_hosts()]}
def append_rule(self, target, source=None, dpt=None, prot='all', chain='INPUT'):
"""
向iptables规则后面新增规则
:param target: str 放行还是屏蔽 指[ACCEPT, DROP]
:param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
:param source: str 源ip地址
:param dpt: str 目的地端口
:param prot: str 协议
:return: tuple (True, None)
"""
if not any([source, dpt]):
return False, 'source 或者 dpt参数缺失'
if dpt is not None and prot not in ['tcp', 'udp']:
return False, 'dpt 参数不能出现在非tcp或者udp协议内'
if source is None and prot in ['tcp', 'udp']:
cmd = 'iptables -A {chain} -p {prot} --dport {dpt} -j {target}'.format(
chain=chain, prot=prot, dpt=dpt, target=target
)
elif all([chain, prot, source, dpt]):
cmd = 'iptables -A {chain} -p {prot} -s {source} --dport {dpt} -j {target}'.format(
chain=chain, prot=prot, source=source, dpt=dpt, target=target
)
else:
cmd = 'iptables -A {chain} -p {prot} -s {source} -j {target}'.format(
chain=chain, prot=prot, source=source, target=target
)
res = self._run_shell_task('append rules', cmd)
if res[self._hosts]['rc'] != 0:
return False, res[self._hosts]['stderr']
else:
res, exception = self._save_rule()
if exception is not None:
return False, exception
return True, None
def delete_rule(self, num, chain='INPUT'):
"""
删除iptables规则
:param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
:param num: str 对应的股则序号id
:return: tuple (True, None)
"""
cmd = 'iptables -D {} {}'.format(chain, num)
res = self._run_shell_task('append rules', cmd)
if res[self._hosts]['rc'] != 0:
return False, res[self._hosts]['stderr']
else:
res, exception = self._save_rule()
if exception is not None:
return False, exception
return True, None
def check_rules(self, num, chain='INPUT'):
"""
检查iptables的规则序号是否正确
:param chain: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
:param num: str 出入站或者转发的目标规则，大些，特指 [INPUT, FORWARD, OUTPUT]三个值
:return: tuple (True, None)
"""
res, exception = self.get_iptables()
if exception is not None:
return False, exception
data = res[chain.lower()]
nums = [i['num'] for i in data]
if num not in nums:
return False, '{}规则内没有num为{}的规则！'.format(chain, num)
else:
return True, None
def search_rules(self, source=None, dpt=None, prot=None, chain='INPUT'):
if not any([source, dpt, prot]):
return None, '缺少查询条件'
data, err = self.get_iptables()
if err is not None:
return None, err
data = data[chain.lower()]
res = list()
for i in data:
if all([source, dpt, prot]):
if source in i['source'] and (i['rules'] is not None and dpt in i['rules']) and prot in i['prot']:
res.append(i)
elif all([source, dpt]):
if source in i['source'] and (i['rules'] is not None and dpt in i['rules']):
res.append(i)
elif all([dpt, prot]):
if dpt in i['source'] and prot in i['prot']:
res.append(i)
elif source:
if source in i['source']:
res.append(i)
elif dpt:
if i['rules'] is not None and dpt in i['rules']:
res.append(i)
elif prot:
if prot in i['prot']:
res.append(i)
return res, None
内容总结
以上是互联网集市为您收集整理的05.Ansible Python API全部内容，希望文章能够帮你解决05.Ansible Python API所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】