利用Pool类多进程实现批量主机管理
1 #!/usr/bin/python
2 # -*- coding: UTF-8 -*-
3 # Author: standby
4 # Time: 2017-03-02
5 # Description: Achieve the Multiple Processes(High Concurrent) to execute single or multiple commands functions by pool class of Python Lang.
6
7 import time
8 import commands, subprocess
9 import os, re, sys
10 import paramiko
11 from multiprocessing import Pool
12
13 # print color
14 COLOR_PINK = ' 33[95m'
15 COLOR_BLUE = ' 33[94m'
16 COLOR_GREEN = ' 33[92m'
17 COLOR_YELLOW = ' 33[93m'
18 COLOR_RED = ' 33[91m'
19 COLOR_DEFAULT = ' 33[0m'
20
21 def Check_IP(ip):
22 ip_str = r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?).){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
23 if re.match(ip_str, ip):
24 return True
25 else:
26 return False
27
28 def Get_IPS(ipfile):
29 '''Generate ip list.'''
30 ips = []
31 with open(ipfile) as iplist:
32 content = iplist.readlines()
33 for line in content:
34 if line.startswith('#'):
35 continue
36 elif Check_IP(line.strip('
')):
37 ips.append(line.strip('
'))
38 else:
39 print '%s is invalid ip address!' % line.strip('
')
40 continue
41 return ips
42
43 def concurrentFunc(ip, cmd):
44 '''single cmd to exec...'''
45 #RET = subprocess.check_output("ssh root@%s 2> /dev/null cmd" % ip, shell = True)
46 #status, output = commands.getstatusoutput("ssh root@%s 2> /dev/null cmd" % ip)
47 #status, output = commands.getstatusoutput("ssh root@%s 2> /dev/null cmd
" % ip)
48 #return ip+": "+output+", status: "+bytes(status)
49 #return ip+": "+RET
50 '''multiple cmd to exec...'''
51 PORT = 22
52 USER = "root"
53 KEY = "/path/known_hosts"
54 ssh = paramiko.SSHClient()
55 try:
56 ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
57 ssh.connect(ip, PORT, USER, KEY, timeout=10)
58 except paramiko.AuthenticationException:
59 print ip+" ssh timeout, continue..."
60 return "SSH ERROR, exit..."
61 output = []
62 output.append(ip)
63 for m in cmd:
64 stdin, stdout, stderr = ssh.exec_command(m)
65 output.append(stdout.readlines())
66 return output
67
68 def callBackFunc(ret):
69 print "This is callback func of %s" % ret[0]
70
71 def output(res_list):
72 print "%s=================RESULT====================%s" % (COLOR_GREEN, COLOR_DEFAULT)
73 for res in res_list:
74 try:
75 print res.get()[0] + " -> " + ''.join(res.get()[1])
76 except Exception, e:
77 print "%sOUTPUT ERROR: %s %s" % (COLOR_YELLOW, e, COLOR_DEFAULT)
78 continue
79
80 if __name__ == '__main__':
81 ipfile = sys.argv[1]
82 if os.path.isfile(ipfile):
83 ips = Get_IPS(ipfile)
84 elif Check_IP(ipfile):
85 ips = [sys.argv[1]]
86 else:
87 print '%s is invalid ip address!' % ipfile
88 sys.exit(1)
89
90 res_list = []
91 #cmd = ['cmd1', 'cmd2']
92 cmd = ['cmd']
93 t_start=time.time()
94 #pool = Pool(32)
95 pool = Pool(10)
96 for ip in ips:
97 #维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
98 res = pool.apply_async(func=concurrentFunc, args=(ip, cmd,), callback=callBackFunc)
99 res_list.append(res)
100 pool.close()
101 pool.join()
102 pool.terminate()
103 output(res_list)
104 t_end=time.time()
105 t=t_end-t_start
106 print '%sDealt %d, used time is :%s.%s' % (COLOR_BLUE, len(res_list), t, COLOR_DEFAULT)