简介
Ansible 在2.0版本后重构了大部分的代码逻辑, 启用了2.0版本之前的 Runner 和 Playbook 类, 使得广大同学之前的代码运行错误。
在ansible1.9的时候，API是一个非常简单的东西。官方说“it's pretty simple”，真是又pretty又simple。到了ansible2.0以后，是“a bit more complicated”，简直让人难受。不过更灵活了
这里我自己对官方的2.0版本之后的 Python API 做了一下优化，做的还不是很好，有问题希望大家能指出。
步骤
1. 测试架构是否可用
-
接着之前的文章继续，首先测试工程是否可用，创建test文件，输入 hello world 。
测试成功
有一个favicon.ico没有访问到，这个忽略。
2. 在服务器端搭建ansible服务
- 使用pip安装ansible
[root@test01 ~]# workon venv301
(venv301) [root@test01 ~]# pip install ansible
(venv301) [root@test01 ~]# pip list
Package Version
------------ -------
ansible 2.4.3.0
asn1crypto 0.24.0
bcrypt 3.1.4
cffi 1.11.5
click 6.7
cryptography 2.1.4
Flask 0.12.2
idna 2.6
itsdangerous 0.24
Jinja2 2.10
MarkupSafe 1.0
paramiko 2.4.0
pip 9.0.1
pyasn1 0.4.2
pycparser 2.18
PyNaCl 1.2.1
PyYAML 3.12
setuptools 38.5.1
six 1.11.0
Werkzeug 0.14.1
wheel 0.30.0
- 创建ansible配置文件
这个也可以不用创建，我这里创建是为了做测试，测试在服务器端ansible可以运行。写代码的时候会把选项写道 python 程序中。
(venv301) [root@test01 ~]# mkdir /etc/ansible
(venv301) [root@test01 venv301]# cat /etc/ansible/ansible.cfg
[defaults]
inventory = /root/test/venv/venv301/hosts
remote_tmp = ~/.ansible/tmp
local_tmp = ~/.ansible/tmp
forks = 15
sudo_user = root
transport = smart
remote_port = 22
host_key_checking = False
deprecation_warnings = False
(venv301) [root@test01 venv301]# cdvirtualenv
(venv301) [root@test01 venv301]# pwd
/root/test/venv/venv301
(venv301) [root@test01 venv301]# cat /root/test/venv/venv301/hosts
[test01]
192.168.1.30
[test01:vars]
ansible_ssh_pass='123456'
- 运行 ansible 测试是否能成功运行
(venv301) [root@test01 venv301]# ansible test01 -m shell -a 'hostname'
192.168.8.30 | FAILED | rc=-1 >>
to use the 'ssh' connection type with passwords, you must install the sshpass program
(venv301) [root@test01 venv301]#
发现缺少 sshpass 模块，下面我们安装后再测试。结果成功了
(venv301) [root@test01 venv301]# yum -y install sshpass
(venv301) [root@test01 venv301]# ansible test01 -m shell -a 'hostname'
192.168.1.30 | SUCCESS | rc=0 >>
test01.localdomain
(venv301) [root@test01 venv301]#
3. 编写测试的 Python 脚本
- 首先客户端的 PyCharm 要关闭重新打开一下，因为刚才我们在服务器端安装了 ansible 。（一直在寻找不用重启 PyCharm ，就能将服务器端安装的软件同步过来的方法，没找到。知道的大神可告诉我，多谢。）
- 参考官方给的文档 Python API 2.0，在test目录下创建测试文件 ansible_test.py，内容如下：
修改了
#!/usr/bin/env python
import json
from collections import namedtuple
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
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))
Options = namedtuple('Options', ['connection', 'module_path', 'forks', 'become', 'become_method', 'become_user', 'check', 'diff'])
# initialize needed objects
loader = DataLoader()
options = Options(connection='smart', module_path=None, forks=100, become=None, become_method=None, become_user=None, check=False,
diff=False)
passwords = dict(vault_pass='secret')
# Instantiate our ResultCallback for handling results as they come in
results_callback = ResultCallback()
# create inventory and pass to var manager
inventory = InventoryManager(loader=loader, sources=['./hosts'])
variable_manager = VariableManager(loader=loader, inventory=inventory)
# create play with tasks
play_source = dict(
name = "Ansible Play",
hosts = 'test01',
gather_facts = 'no',
tasks = [
dict(action=dict(module='shell', args='hostname'), register='shell_out'),
dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# actually run it
tqm = None
try:
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=variable_manager,
loader=loader,
options=options,
passwords=passwords,
stdout_callback=results_callback, # Use our custom callback instead of the ``default`` callback plugin
)
result = tqm.run(play)
finally:
if tqm is not None:
tqm.cleanup()
-
创建要测试的 Inventory （hosts.txt 文件）
inventory文件
-
运行 ansible_test.py 验证执行成功
python的ansible执行成功
4. 改进官方给的 ansible Python API
- 这里我单创建了一个 ansible 文件夹，用来存放相关的文件。改进后的文件内容如下：
#!/usr/bin/env python
import json
from collections import namedtuple
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
class ResultCallback(CallbackBase):
def __init__(self):
self.x = dict()
self.y = dict()
self.z = dict()
def v2_runner_on_ok(self, result, **kwargs):
host = result._host
# 执行成功后的结果放到x字典中
self.x[host.name] = result._result['stdout']
def v2_runner_on_failed(self, result, ignore_errors=True):
host = result._host
# 执行后的结果放到x失y字典中
self.y[host.name] = result._result['stderr_lines']
def v2_runner_on_unreachable(self, result):
host = result._host
# 后的网络不结果放到x通z字典中
self.z[host.name] = result._result['msg']
class AnsibleRunner:
def __init__(self, remote_user='liuxin', conn_pass='1234567', group='test01', module_name='shell',
module_args='hostname'):
# 在remote服务器上执行命令的用户
self.remote_user = remote_user
# 执行命令用户的密码
self.conn_pass = conn_pass
# 执行命令的主机，可以使group例如test01，也可以使列表例如['192.168.1.30']
self.group = group
# 执行的模块，例如shell,command,ping
self.module_name = module_name
# 命令，例如hostname,whoami
self.module_args = module_args
def order_run(self):
Options = namedtuple('Options',
['connection', 'module_path', 'forks', 'become', 'become_method', 'become_user',
'remote_user', 'check', 'diff'])
# initialize needed objects
loader = DataLoader()
options = Options(connection='smart', module_path=None, forks=100, become=None, become_method=None,
become_user=None, remote_user=self.remote_user, check=False, diff=False)
passwords = dict(vault_pass='secret', conn_pass=self.conn_pass)
# Instantiate our ResultCallback for handling results as they come in
results_callback = ResultCallback()
# create inventory and pass to var manager
inventory = InventoryManager(loader=loader, sources=['./hosts'])
variable_manager = VariableManager(loader=loader, inventory=inventory)
# create play with tasks
play_source = dict(
name="Ansible Play",
hosts=self.group,
gather_facts='no',
tasks=[
dict(action=dict(module=self.module_name, args=self.module_args), register='shell_out'),
# dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}')))
]
)
play = Play().load(play_source, variable_manager=variable_manager, loader=loader)
# actually run it
tqm = None
try:
tqm = TaskQueueManager(
inventory=inventory,
variable_manager=variable_manager,
loader=loader,
options=options,
passwords=passwords,
stdout_callback=results_callback, # Use our custom callback instead of the ``default`` callback plugin
)
result = tqm.run(play)
finally:
if tqm is not None:
tqm.cleanup()
# 这里可以再进一步改进的，实现实时的现实执行结果
return [results_callback.x, results_callback.y, results_callback.z]
# a = AnsibleRunner(remote_user='liuxin', conn_pass='1234567', group='test01', module_name='shell', module_args='whoami')
# print(a.order_run())
#
-
运行测试
成功，结果如下：
测试成功
5. 编写 Flask 的API
- 创建一个简单的 api
具体可以访问我的 github
ansible_web -
测试 api 是否可用
结果如下：
成功
- 加入 ansible 的 api
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2018/3/4 17:15
# @Author : Administrator
# @Site :
# @File : api_1_0_liuxin.py
# @Software: PyCharm
from ..api_1_0 import api_1_0
from flask_restful import Api, Resource
from ..ansible.models import AnsibleRunner
from flask import request, jsonify
my_api = Api(api_1_0)
class HelloWorld(Resource):
def get(self):
return jsonify({'hello': 'world'})
class AnsibleRunnerApi(Resource):
def post(self):
hosts = request.form.getlist('hosts')
# hosts = request.form.get('hosts')
username = request.form['username']
passwd = request.form['passwd']
module_name = request.form['module_name']
module_args = request.form['module_args']
print(hosts, username, passwd, module_name, module_args)
a = AnsibleRunner(remote_user=username, conn_pass=passwd, group=hosts, module_name=module_name,
module_args=module_args)
res = a.order_run()
print(res)
return jsonify(res)
my_api.add_resource(HelloWorld, '/test_hello', endpoint='test_hello')
my_api.add_resource(AnsibleRunnerApi, '/test_api', endpoint='test_api')
-
测试
成功
补充
1. 给创建的 Python 文件自动添加文件头
设置方法如下：
pycharm --- file --- setting --- editor --- file and code templates --- python script
# 可以根据自己的实际情况进行更改，创建文件头如下：
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : ${DATE} ${TIME}
# @Author : ${USER}
# @Site : ${SITE}
# @File : ${NAME}.py
# @Software: ${PRODUCT_NAME}
文件头设置位置截图
网友评论