【实战小项目】python开发自动化运维工具--批量操作主机，,有很多开源自动化运维
【实战小项目】python开发自动化运维工具--批量操作主机，,有很多开源自动化运维
有很多开源自动化运维工具都很好用如ansible/salt stack等，完全不用重复造轮子。只不过，很多运维同学学习Python之后，苦于没小项目训练，本篇演示用Python写一个批量操作主机的工具，大家空余时候可以试着写写，完善完善。
1 思路分析
在运维工作中，古老的方式部署环境、上线代码可能都需要手动在服务器上敲命令，不胜其烦。所以，脚本，自动化工具等还是很有必要的。我觉得一个批量操作工具应该考虑以下几点：
（1）本质上，就是到远程主机上执行命令并返回结果。
（2）做到批量。也就是要并发对多台机器进行操作。
（3）将返回的结果，清晰地展示给用户。
通常开源的主机批量管理工具有两类，一类是有agent，如SaltStack、Puppet等；另一类是无agent如ansible。虽然我们没必要重复造轮子，但是可以试着写一写，一是加深对这类软件原理的理解，二是练习Python。建议如果服务器规模在1000台以内的用无agent的方式也能hold住；如果超过1000台，用有agent的会好太多。
接下来我们一起看看怎么具体实现。
2到远程机器上执行命令
到远程机器上执行命令，并返回结果，至少有两种方式：一是用paramiko模块；而是可以建立机器互信，从中控执行ssh命令。
下面我把自己封装好的代码贴一下，是基于paramiko模块封装的，ssh的大家可以自己实现：
import paramikoclass SSHParamiko(object): err = "argument passwd or rsafile can not be None" def __init__(self, host, port, user, passwd=None, rsafile=None): self.h = host self.p = port self.u = user self.w = passwd self.rsa = rsafile def _connect(self): if self.w: return self.pwd_connect() elif self.rsa: return self.rsa_connect() else: raise ConnectionError(self.err) def _transfer(self): if self.w: return self.pwd_transfer() elif self.rsa: return self.rsa_transfer() else: raise ConnectionError(self.err) def pwd_connect(self): conn = paramiko.SSHClient() conn.set_missing_host_key_policy(paramiko.AutoAddPolicy()) conn.connect(self.h, self.p, self.u, self.w) return conn def rsa_connect(self): pkey = paramiko.RSAKey.from_private_key_file(self.rsa) conn = paramiko.SSHClient() conn.set_missing_host_key_policy(paramiko.AutoAddPolicy()) conn.connect(hostname=self.h, port=self.p, username=self.u, pkey=pkey) return conn def pwd_transfer(self): transport = paramiko.Transport(self.h, self.p) transport.connect(username=self.u, password=self.w) sftp = paramiko.SFTPClient.from_transport(transport) return sftp, transport def rsa_transfer(self): pkey = paramiko.RSAKey.from_private_key_file(self.rsa) transport = paramiko.Transport(self.h, self.p) transport.connect(username=self.u, pkey=pkey) sftp = paramiko.SFTPClient.from_transport(transport) return sftp, transport def run_cmd(self, cmd): conn = self._connect() stdin, stdout, stderr = conn.exec_command(cmd) code = stdout.channel.recv_exit_status() stdout, stderr = stdout.read(), stderr.read() conn.close() if not stderr: return code, stdout.decode() else: return code, stderr.decode() def get_file(self, remote, local): sftp, conn = self._transfer() sftp.get(remote, local) conn.close() def put_file(self, local, remote): sftp, conn = self._transfer() sftp.put(local, remote) conn.close()
当然，代码还可以重构一下哈。接下来我们看下效果：
if __name__ == ‘__main__‘: h = "我的测试机IP" p = 22 u = "我的用户名" w = "我的密码" obj = SSHParamiko(h, p, u, w) r = obj.run_cmd("df -h") print(r[0]) print(r[1]) 执行之后的结果是：
0Filesystem Size Used Avail Use% Mounted on/dev/vda1 40G 3.4G 34G 9% /devtmpfs 3.9G 0 3.9G 0% /devtmpfs 3.9G 0 3.9G 0% /dev/shmtmpfs 3.9G 410M 3.5G 11% /runtmpfs 3.9G 0 3.9G 0% /sys/fs/cgroup/dev/vdb 300G 12G 289G 4% /search/odintmpfs 783M 0 783M 0% /run/user/0tmpfs 783M 0 783M 0% /run/user/1000
可以清晰看到第一行是命令执行的状态码，0表示成功，非0表示失败；第二行开始就是我们的命令返回结果。是不是比较清晰呢？
3 并发执行并展示输出结果
并发执行通常用Python3自带的线程模块就行，这里我用的from concurrent.futures import ThreadPoolExecutor。并且当拿到结果之后，我还做了一些格式化输出，比如绿色输出表示成功，红色输出表示命令执行失败，黄色表示提醒等。废话不多说，直接看代码吧！
from concurrent.futures import ThreadPoolExecutorclass AllRun(object): def __init__(self, ssh_objs, cmds, max_worker=50): self.objs = [o for o in ssh_objs] self.cmds = [c for c in cmds] self.max_worker = max_worker # 最大并发线程数 self.success_hosts = [] # 存放成功机器数目 self.failed_hosts = [] # 存放失败的机器IP self.mode = None self.func = None def serial_exec(self, obj): """单台机器上串行执行命令，并返回结果至字典""" result = list() for c in self.cmds: r = obj.run_cmd(c) result.append([c, r]) return obj, result def concurrent_run(self): """并发执行""" future = ThreadPoolExecutor(self.max_worker) for obj in self.objs: try: future.submit(self.serial_exec, obj).add_done_callback(self.callback) except Exception as err: err = self.color_str(err, "red") print(err) future.shutdown(wait=True) def callback(self, future_obj): """回调函数，处理返回结果""" ssh_obj, rlist = future_obj.result() print(self.color_str("{} execute detail:".format(ssh_obj.h), "yellow")) is_success = True for item in rlist: cmd, [code, res] = item info = f"{cmd} | code => {code}\nResult:\n{res}" if code != 0: info = self.color_str(info, "red") is_success = False if ssh_obj.h not in self.failed_hosts: self.failed_hosts.append(ssh_obj.h) else: info = self.color_str(info, "green") print(info) if is_success: self.success_hosts.append(ssh_obj.h) if ssh_obj.h in self.failed_hosts: self.failed_hosts.remove(ssh_obj.h) def overview(self): """展示总的执行结果""" for i in self.success_hosts: print(self.color_str(i, "green")) print("-" * 30) for j in self.failed_hosts: print(self.color_str(j, "red")) info = "Success hosts {}; Failed hosts {}." s, f = len(self.success_hosts), len(self.failed_hosts) info = self.color_str(info.format(s, f), "yellow") print(info) @staticmethod def color_str(old, color=None): """给字符串添加颜色""" if color == "red": new = "\033[31;1m{}\033[0m".format(old) elif color == "yellow": new = "\033[33;1m{}\033[0m".format(old) elif color == "blue": new = "\033[34;1m{}\033[0m".format(old) elif color == "green": new = "\033[36;1m{}\033[0m".format(old) else: new = old return newif __name__ == ‘__main__‘: h1 = "adime01.shouji.sjs.ted" p1 = 22 u1 = "odin" w1 = "*****" h = "10.129.206.97" p = 22 u = "root" w = "*****" obj1 = SSHParamiko(h1, p1, u1, w1) obj = SSHParamiko(h, p, u, w) cmds = ["df -h", "ls"] all_obj = AllRun([obj1, obj], cmds) all_obj.concurrent_run() all_obj.overview()上述代码运行的结果：
从执行结果来看，高亮显示，清新明了。既显示了各个主机的各个命令执行状态码，返回结果，最后还汇总结果，成功了多少台机器和失败了多少台机器。
我们还可以换一下执行的命令，让命令执行失败看看：
后期还可以包装一下，将主机、密码、批量执行的命令写在配置文件中；或再根据需要包装成命令行工具，在日常运维工作中可以适当减少人肉敲命令的繁琐。
【实战小项目】python开发自动化运维工具--批量操作主机
评论关闭