Ansible AdHoc & playbook API + 动态生成Inventory +结果关注

标签：性能   self   parse   let   ogg   源码   agent   each   book   

直接贴代码，没太多注释，看不懂的看下源码。Pycharm+b

Ansible 2.0 之后的 API 比 2.0 之前要复杂，但使用起来的自由度更好，可根据自己需求修改 Ansible API 的使用方法；功能也更强大。

我主要是使用这个 API 配合 Djcelery 实现监控系统的数据采集功能，好处是不再需要每中服务器再开发一个agent。这样使用的问题可能主要是程序的负载性能及程序并发性能，我还没测试。

-- coding:utf8 --

import os
import sys
import logging
logger = logging.getLogger(‘django‘)

from collections import namedtuple

from ansible.inventory import Inventory
from ansible.vars import VariableManager
from ansible.parsing.dataloader import DataLoader
from ansible.inventory.group import Group
from ansible.inventory.host import Host
from ansible.playbook.play import Play
from ansible.executor.playbook_executor import PlaybookExecutor
from ansible.executor.task_queue_manager import TaskQueueManager
from ansible.plugins.callback import CallbackBase

OPTIONS = namedtuple(‘OPTIONS‘,

[‘module_path‘, ‘extra_vars‘, ‘forks‘, ‘become‘, ‘become_method‘, ‘become_user‘,

‘become_ask_pass‘, ‘connection‘, ‘timeout‘, ‘poll_interval‘, ‘check‘, ‘diff‘]

)

options = OPTIONS(module_path=None, extra_vars=None, forks=5, become=None, become_method=None, become_user=None,

become_ask_pass=None, connection=‘smart‘, timeout=10, poll_interval=15, check=False, diff=3)

class Options(object):

"""
Initialize options class to replace Ansible OptParser
"""

def __init__(self, module_path=None, extra_vars=None, forks=10, become=None, become_method=None, become_user=None,
             become_ask_pass=None, connection=‘smart‘, timeout=2, poll_interval=15, check=False, diff=3,
             listtasks=None, listhosts=None, listtags=None, syntax=None):
    self.module_path = module_path
    self.extra_vars = extra_vars
    self.forks = forks
    self.become = become
    self.become_method = become_method
    self.become_user = become_user
    self.become_ask_pass = become_ask_pass
    self.connection = connection
    self.timeout = timeout
    self.poll_interval = poll_interval
    self.check = check
    self.diff = diff
    self.listhosts = listhosts
    self.listtasks = listtasks
    self.listtags = listtags
    self.syntax = syntax

class ResultsCallback(CallbackBase):

‘‘‘
Callback the result of execute AdHoc and playbook
‘‘‘

def __init__(self, *args, **kwargs):
    super(ResultsCallback, 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 v2_runner_on_async_poll(self, result):
    self.host_ok.setdefault(‘async_poll‘, result)

def v2_runner_on_async_ok(self, result):
    self.host_ok.setdefault(‘async_ok‘, result)

def v2_runner_on_async_failed(self, result):
    self.host_failed.setdefault(‘async_failed‘, result)

class DynamicInventory(Inventory):

def __init__(self, resource, loader, variable_manager):

    ‘‘‘
    @resource:
    { 
        "group1": { 
            "hosts": [{"hostname": "10.0.0.0", "port": "22", "username": "test", "password": "pass"}, ...], 
            "vars": {"var1": value1, "var2": value2, ...} 
        } 
    }
    ‘‘‘

    self.resource = resource
    self.inventory = Inventory(loader=loader, variable_manager=variable_manager, host_list=[])
    self.gen_inventory()

def Dynamic_add_group(self, hosts, groupname, groupvars=None):
    NewGroup = Group(name=groupname)

    # if group variables exists, add them to group
    if groupvars:
        for key, value in groupvars.iteritems():
            NewGroup.set_variable(key, value)

    # add hosts to group
    for host in hosts:
        # set connection variables
        hostname = host.get("hostname")
        hostport = host.get("port")
        username = host.get("username")
        password = host.get("password")
        # ssh_key = host.get("ssh_key")
        GeneralHost = Host(name=hostname, port=hostport)
        GeneralHost.set_variable(‘ansible_ssh_host‘, hostname)
        GeneralHost.set_variable(‘ansible_ssh_port‘, hostport)
        GeneralHost.set_variable(‘ansible_ssh_user‘, username)
        GeneralHost.set_variable(‘ansible_ssh_pass‘, password)
        # GeneralHost.set_variable(‘ansible_ssh_private_key_file‘, ssh_key)

        # set other variables
        for key, value in host.iteritems():
            if key not in ["hostname", "port", "username", "password"]:
            # if key not in ["hostname", "port"]:
                GeneralHost.set_variable(key, value)
        # add to group
        NewGroup.add_host(GeneralHost)

    return self.inventory.add_group(NewGroup)

def gen_inventory(self):

    """
    Dynamic add host to inventory.
    """

    if isinstance(self.resource, list):
        self.Dynamic_add_group(self.resource, ‘default_group‘)
    elif isinstance(self.resource, dict):
        for groupname, hosts_and_vars in self.resource.iteritems():
            self.Dynamic_add_group(hosts_and_vars.get("hosts"), groupname, hosts_and_vars.get("vars"))

class AnsibleAPI(object):

def __init__(self, resource):

    ‘‘‘
    @resource type: dict
    { 
        "group1": { 
            "hosts": [{"hostname": "10.0.0.0", "port": "22", "username": "test", "password": "pass"}, ...], 
            "vars": {"var1": value1, "var2": value2, ...} 
        } 
    }
    @resource: list
    { 
        "hosts": [{"hostname": "10.0.0.0", "port": "22", "username": "test", "password": "pass"}
    }
    ‘‘‘
    self.variable_manager = VariableManager()
    self.loader = DataLoader()
    self.options = Options()
    self.passwords = dict()
    self.results_raw = {‘success‘: {}, ‘failed‘: {}, ‘unreachable‘: {}}
    self.resource = resource
    self.callback = ResultsCallback()
    self.inventory = DynamicInventory(self.resource, self.loader, self.variable_manager).inventory
    self.variable_manager.set_inventory(self.inventory)

    if isinstance(self.resource, list):
        self.host_list = map(str, self.inventory.get_hosts(pattern=‘default_group‘))

    if isinstance(self.resource, dict):
        self.host_list = []
        for groupname in self.resource.keys():
            self.host_list.extend(self.inventory.get_group(groupname).get_hosts())
        self.host_list = map(str, self.host_list)

def PrivateAdHoc(self, module_name, module_args=‘‘):

    """
    run module from andible ad-hoc.
    module_name: ansible module_name
    module_args: ansible module args
    """

    # create play with tasks
    play_source = dict(
            name="Ansible Play",
            hosts=self.host_list,
            gather_facts=‘no‘,
            tasks=[dict(action=dict(module=module_name, args=module_args))]
    )
    play = Play().load(play_source, variable_manager=self.variable_manager, loader=self.loader)

    # Actually run it
    tqm = None
    try:
        tqm = TaskQueueManager(
                inventory=self.inventory,
                variable_manager=self.variable_manager,
                loader=self.loader,
                options=self.options,
                passwords=self.passwords,
                stdout_callback=self.callback
        )
        result = tqm.run(play)
    finally:
        if tqm is not None:
            tqm.cleanup()

    return result

def PrivatePlaybook(self, playbooks):

    """
    run ansible palybook and get ran result
    """

    try:
        if not os.path.exists(playbooks):
            logger.error(‘No such file: %s‘ % playbooks)
            sys.exit()
        self.executor = PlaybookExecutor(playbooks=[playbooks, ],
                                         inventory=self.inventory,
                                         variable_manager=self.variable_manager,
                                         loader=self.loader,
                                         options=self.options,
                                         passwords=self.passwords
                                         )

        print self.inventory.get_group_dict()

        self.executor._tqm._stdout_callback = self.callback
        result = self.executor.run()

    except Exception as e:
        print "Error: %s" % e
    finally:
        return result

def playbook_result(self):

    for host, result_object in self.executor._tqm._stdout_callback.host_ok.iteritems():
        self.results_raw[‘success‘][host] = result_object._result

    for host, result_object in self.executor._tqm._stdout_callback.host_failed.iteritems():
        self.results_raw[‘failed‘][host] = result_object._result

    for host, result_object in self.executor._tqm._stdout_callback.host_unreachable.iteritems():
        self.results_raw[‘failed‘][host] = result_object._result

    return self.results_raw

def AdHoc_result(self):

    for host, result in self.callback.host_ok.items():
        self.results_raw[‘success‘][host] = result._result

    for host, result in self.callback.host_failed.items():
        try:
            self.results_raw[‘failed‘][host] = result._result[‘stderr‘]
        except KeyError:
            self.results_raw[‘failed‘][host] = result._result[‘msg‘]

    for host, result in self.callback.host_unreachable.items():
        self.results_raw[‘unreachable‘][host] = result._result[‘msg‘]

    return self.results_raw

if name == ‘main‘:

print ‘complete‘
    #### 以下是测试成功示例，主要关注如何传参
# example = AnsibleAPI([
    # dict(hostname=‘10.10.181.132‘, port=20003, username=‘root‘, password=123456),
# ])
# example.PrivateAdHoc(‘service‘, module_args=‘name=crond state=reloaded‘)
# example.PrivateAdHoc(‘setup‘)
# print example.AdHoc_result()
# example = AnsibleAPI(dict(memory=dict(hosts=[dict(hostname=‘10.10.181.132‘, port=20003, username=‘root‘, password=123456)], vars=dict())))
# print example.PrivatePlaybook(‘/Users/fanolee/PycharmProjects/AutoOPPlatform/monitor/ansible/playbooks/entry_files/memory/memory.yml‘)
# print example.playboot_result()

Ansible AdHoc & playbook API + 动态生成Inventory +结果关注

标签：性能   self   parse   let   ogg   源码   agent   each   book   

原文地址：http://blog.51cto.com/784687488/2171821