常用模块
用于读取yaml，json格式的文件 from ansible.parsing.dataloader import DataLoader #用于管理变量的类，包括主机，组，扩展等变量 from ansible.vars.manager import VariableManager #用于创建和管理inventory，倒入inventory文件 from ansible.inventory.manager import InventoryManager #ad-hoc 存储着角色列表，任务，处理代码块 from ansible.playbook.play import Play #ad-hoc ansible底层用到的任务队列 from ansible.executor.task_queue_manager import TaskQueueManager #回调基类，用来定义回调事件，比如返回失败成功等信息 from ansible.plugins.callback import CallbackBase #执行playbook from ansible.executor.playbook_executor import PlaybookExecutor #操作单个主机 from ansible.inventory import host #操作单个主机组 from ansible.inventory import group
InventoryManager
#实例化需要两个参数 "参数一为读取yml文件的信息，需要实例化 DataLoader" "参数二为读取从那个位置读取资产配置文件，多个可逗号分割" intertory = InventoryManager(loader='',sources='') #以字典的方式打印出主机和主机组相关信息 intertory.get_group_dict() #获取所有的主机 inventory.get_hosts() #添加主机到指定主机组 "参数一指定主机地址" "参数二指定主机端口" "参数三指定主机群组，必须存在的群组" inventory.add_host(host='1.1.1.1',port=2222,group='web_server') #获取指定的主机对象 inventory.get_host(hostname='1.1.1.1')
VariableManager
#实例化需要两个参数 "参数一为读取yml文件的信息，需要实例化 DataLoader" "参数二为资产管理配置变量" variable = VariableManager(loader=loader,inventory=inventory) #获取变量 variable.get_vars() # 查看指定主机的详细变量信息 "传入的host是inventory.get_host获得的主机对象" host = inventory.get_host(hostname='1.1.1.1') host_vars = variable.get_vars(host=host) #设置主机变量方法 "传入的host是inventory.get_host获得的主机对象" host = inventory.get_host(hostname='1.1.1.1') variable.set_host_variable(host=host,varname='ansible_ssh_pass',value='12345') #添加扩展变量 "参数是一个字典多个逗号分割" variable.extra_vars={'devops':'best'}
基于CMDB接口返回主机&主机信息进行动态执行
#!/usr/bin/env python # -*- coding:utf-8 -*- import json import sys import os import django sys.path.append('../../') os.environ.setdefault("DJANGO_SETTINGS_MODULE", "automation_ansible_api.settings") django.setup() from assets.models import Group def Group_list(): ''' 这个函数的目的是获取所有主机以及群组的相关信息，必须是json格式,格式如下 { "Devops": { "hosts": [ "172.20.35.177", "172.20.35.16", "106.14.154.11", "139.196.90.14", "140.143.191.129" ] } } ''' result = {} groups = Group.objects.all() #主机组的所有名字，多对多关系这里能反向查找出来 for i in groups: if i.host_set.all(): result[i.name] = {'hosts':[]} for x in i.host_set.all(): result[i.name]['hosts'].append(x.ip_address) print(json.dumps(result,indent=4)) def Host_list(ip): ''' 这个函数目的是存储一些变量，inventory的一些参数比如执行的用户，密码，端口等等，可返回空，根据自身需求 { "ansible_ssh_host": "172.20.35.177", "ansible_ssh_user": "root" } ''' print(json.dumps({ "ansible_ssh_host": ip, "ansible_ssh_user": "root"},indent=4)) if __name__ == '__main__': ''' 一下两个参数必须存在，这个是用于ansible去执行和发现的！ ''' if len(sys.argv) == 2 and (sys.argv[1] == '--list'): Group_list() elif len(sys.argv) == 3 and (sys.argv[1] == '--host'): Host_list(sys.argv[2])
ansible-api调用执行
#!/usr/bin/env python # -*- coding:utf-8 -*- import json from ansible.plugins.callback import CallbackBase 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 collections import namedtuple #实例化解析yml loader = DataLoader() #实例化资产管理 ''' 2.4+的inventory必须有sources参数，参数路径为动态执行的那个脚本！这里很坑我研究了一天！ ''' inventory = InventoryManager(loader=loader,sources='hosts') #实例化变量管理 variable_manager = VariableManager(loader=loader,inventory=inventory) 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 = 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_source = dict( ''' hosts为执行的主机或者群组！ ''' name = "Ansible Play ad-hoc test", hosts = '139.196.90.14', gather_facts = 'no', tasks = [ dict(action=dict(module='ping', args='')), # dict(action=dict(module='debug', args=dict(msg='{{shell_out.stdout}}'))) ] ) play = Play().load(play_source, variable_manager=variable_manager, loader=loader) class ModelResultsCollector(CallbackBase): def __init__(self, *args, **kwargs): super(ModelResultsCollector, 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): self.host_ok[result._host.get_name()] = result def v2_runner_on_failed(self, result): self.host_failed[result._host.get_name()] = result callback = ModelResultsCollector() passwords = dict() # 传入自定义的callback tqm = TaskQueueManager( inventory=inventory, variable_manager=variable_manager, loader=loader, options=options, passwords=passwords, stdout_callback=callback, ) result = tqm.run(play) # 自定义格式化输出，执行结果数据在result对象的_result属性里 result_raw = {'success':{},'failed':{},'unreachable':{}} for host,result in callback.host_ok.items(): result_raw['success'][host] = result._result for host,result in callback.host_failed.items(): result_raw['failed'][host] = result._result print(json.dumps(result_raw,indent=4))