Ansible 多机文件分发、执行脚本并单机合并实验结果（Check point， 多线程异步执行，主机状态检测等）

简介

Ansible其实是一个自动化管理工具，可以管理多台机器的自动化部署与安装。在python中，结合Ansible来试验任务多机执行功能，采用Master-Slave架构，首先检测网络状态，剔除不可用主机，然后进行负载均衡，然后实现数据集的分发与异步执行，在执行过程中，采用check point的机制检测是否执行成功，当全部执行完毕后，Master收集所有实验结果。

Host（host.ini）

[master]
netlab_x2

[slave]
netlab_x1

Config（Part）(config.ini)

[ansible]
master_load = 5
slave_load = 1
code_home = /home/guangfa/ansible/guangfa

Python Script

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import sys, os
import ConfigParser
import ansible.runner
import time

inventory_file = "../host.ini"
database_path = "../database"
config_file = "../config.ini"

class DistributedByAnsible:
    def __init__(self):
        cf = ConfigParser.ConfigParser()
    cf.read(config_file)
    self.master_load = int(cf.get("ansible", "master_load"))
        self.slave_load = int(cf.get("ansible", "slave_load"))

    self.code_home_path = cf.get("ansible", "code_home")
    self.ansibleResult = self.code_home_path + "/ansibleResult"
    self.database_path = self.code_home_path + "/database"


    self.avail_master = []
    self.avail_slave = []
    self.loadBalancing = {}

    def extractHost(self,config_file_path):
        cf = ConfigParser.ConfigParser(allow_no_value = True)
        cf.read(config_file_path)
        s = cf.sections()
        #print 'section:', s
        ms = cf.options("master")
        #print 'options:', ms
    for x in ms:
        runner = ansible.runner.Runner(
        host_list=inventory_file,
            module_name='ping',
        module_args='',
        pattern=x
        )
        res = runner.run()
        #print res
        if res['dark'] and res['dark'][x] and res['dark'][x]['failed']:
        print "Host Error:", x, "Unreachable"
            continue
        self.avail_master.append(x)
    self.avail_master = list(set(self.avail_master))
    print self.avail_master
    ss = cf.options("slave")
    for x in ss:
        runner = ansible.runner.Runner(
            host_list=inventory_file,
            module_name='ping',
        module_args='',
        pattern=x
        )
        res = runner.run()
        #print res
        if res['dark'] and res['dark'][x] and res['dark'][x]['failed']:
            continue
        self.avail_slave.append(x)
    self.avail_slave = list(set(self.avail_slave))

    # make true the host is unique
    for host in self.avail_master:
        if host in self.avail_slave:
            self.avail_slave.remove(host)
    print self.avail_slave
    
    def calculateLoadBalancing(self):
        load_count = int(os.popen("ls %s | grep ^mygraph | wc -l" % database_path).read())
    self.load_count = load_count
    current_load_count = load_count
    print load_count
    print self.master_load, self.slave_load
    master_all_count = len(self.avail_master)
    slave_all_count = len(self.avail_slave)
    
    for m in self.avail_master:
        temp_count = min(current_load_count, int(load_count * (self.master_load / float(self.master_load * master_all_count + self.slave_load * slave_all_count))))
        print temp_count
        if temp_count > 0:
                self.loadBalancing[m] = temp_count
        current_load_count -= temp_count
        for s in self.avail_slave:
        temp_count = min(current_load_count, int(load_count * (self.slave_load / float(self.master_load * master_all_count + self.slave_load * slave_all_count))))
        if temp_count > 0:
            self.loadBalancing[s] = temp_count
        current_load_count -= temp_count

    if current_load_count > 0 and master_all_count > 0:
        self.loadBalancing[self.avail_master[0]] += current_load_count
        current_load_count = 0
    if current_load_count > 0 and slave_all_count > 0:
        self.loadBalancing[self.avail_slave[0]] += current_load_count
        current_load_count = 0
    print self.loadBalancing
    
    def fileTransfer(self):
        print self.code_home_path
    print self.ansibleResult
    if self.ansibleResult == "":
        print "ERROR: the home of the result about ansible is null!"
        sys.exit(1)
    current_database_count = 0
    #clean the directory of the result
    for host in self.loadBalancing:
        print host
        runner = ansible.runner.Runner(
            host_list=inventory_file,
        module_name='file',
        module_args='path=%s state=absent' % self.ansibleResult,
        pattern=host
        )
        res = runner.run()
        #print res
        runner = ansible.runner.Runner(
            host_list=inventory_file,
        module_name='file',
        module_args='path=%s state=directory mode=0750' % self.ansibleResult,
        pattern=host
        )
        res = runner.run()
        #print res
            
        # transfer the database
            this_host_load_count = self.loadBalancing[host]
        for i in range(this_host_load_count) :
            runner = ansible.runner.Runner(
            host_list=inventory_file,
            module_name='copy',
            module_args='src=%s/mygraph%s dest=%s mode=0750' % (self.database_path, current_database_count + i, self.ansibleResult),
            pattern=host
        )
        res = runner.run()
        print res
        #print current_database_count
        #print this_host_load_count
        current_database_count += this_host_load_count

        # transfer the code
        file_code_list = ["community.jar", "config.ini", "runAll.sh","lib"]
        for f in file_code_list:
            runner = ansible.runner.Runner(
            host_list=inventory_file,
            module_name='copy',
            module_args='src=%s/%s dest=%s mode=770' % (self.code_home_path, f, self.code_home_path),
            pattern=host
        )
        res = runner.run()
        print res
             
        # run the code distributed
        runner = ansible.runner.Runner(
            host_list=inventory_file,
        module_name="shell",
        module_args='chdir=%s nohup bash runAll.sh %s > bash.log &' % (self.code_home_path, self.ansibleResult),
        pattern=host
        )

        res = runner.run()
        print res
        print "The Host named", host, "is running!"

    def informationTransfer(self, transfer_module_name, transfer_module_args, host):
        print transfer_module_args
    print transfer_module_name
    runner = ansible.runner.Runner(
        host_list=inventory_file,
        module_name=transfer_module_name,
        module_args=transfer_module_args,
        pattern=host
    )
    res = runner.run()
    print res

    def checkPoint(self):
        os.system("rm -rf %s/output" % self.database_path)
        os.system("mkdir -p %s/output" % self.database_path)
        print "Begining CheckPoint"
    host_keys = self.loadBalancing.keys()
    #print host_keys
    while len(host_keys) > 0:
        for host in host_keys:
        runner = ansible.runner.Runner(
            host_list=inventory_file,
            module_name='command',
            module_args='removes=%s/success_log pwd' % self.ansibleResult,
            pattern=host
        )
        res = runner.run()
        print res
        if res['contacted'] and res['contacted'][host] and "skipped" in res['contacted'][host]['stdout']:
            continue
        else:
            print "Host", host, "finished his tasks!"
            host_keys.remove(host)
            self.informationTransfer("synchronize", "src=%s/output dest=%s mode=pull" % (self.ansibleResult, self.database_path), host)
            print host_keys
            print "to sleep"
            time.sleep(5)
    

if __name__ == "__main__":
    ob = DistributedByAnsible()
    ob.extractHost(inventory_file)
    #ob.calculateLoadBalancing();
    #ob.fileTransfer();
    #ob.checkPoint();
    print "Done"

    

待完善部分

主机检测状态仅在开始时检测，剔除无用主机，但是当脚本运行中途机器死掉，此时需要分发另一台主机来执行任务。