Python RPC 远程调用脚本之 RPyC 实践

最近有个监控需求，需要远程执行集群每个节点上的脚本，并获取脚本执行结果，为了安全起见不需要账号密码登陆节点主机，要求只需要调用远程脚本模块的方法就能实现。

总结下python进行远程调用脚本方法：

• 登陆主机执行脚本，python模块支持如 pssh、pexpect、paramiko、ansible

• 以远程方法调用（不需要登陆主机），python模块 rpyc，支持分布式

• socket 方式，稍显复杂，需要熟悉网络协议，起点比较高

rpyc支持远程调用、分布式计算，以较少代码量实现复杂socket编程，本文主要介绍 rpyc 并用它来实现一个 demo。

以代码方式介绍:

需求：分别执行集群每个节点上 server 端的脚本，并返回执行结果给 client 端

1Monitor_RPC_Client.py
2
3#!/usr/bin/env python
4# coding=utf-8
5# 测试utf-8编码
6# python exec_cmd.py "ls -lrt /opt/data1/logs/nginx/pc/track/`date +'%Y%m%d'`|awk '{s+=\$5}END{print s}'"
7# python exec_cmd.py "wc -l /opt/data1/logs/nginx/pc/track/`date +'%Y%m%d'`/*|awk '{s+=\$1}END{print s}'"
8import sys
9
10reload(sys)
11sys.setdefaultencoding('utf-8')
12
13import rpyc
14from pyUtil import *
15from multiprocessing.dummy import Pool as ThreadPool
16
17hostDict = {
18    '192.168.1.216': 12345,
19    '192.168.1.217': 12345,
20    '192.168.1.218': 12345
21}
22
23localResultDict = {}
24
25
26def rpc_client(host_port_cmd):
27    host = host_port_cmd[0]
28    port = host_port_cmd[1]
29    cmd = host_port_cmd[2]
30    c = rpyc.connect(host, port)
31    result = c.root.exposed_execCmd(cmd)
32    localResultDict[host] = result
33    c.close()
34
35
36def exec_cmd(cmd_str):
37    host_port_list = []
38    for (host, port) in hostDict.items():
39        host_port_list.append((host, port, cmd_str))
40
41    pool = ThreadPool(len(hostDict))
42    results = pool.map(rpc_client, host_port_list)
43    pool.close()
44    pool.join()
45    for ip, result in sorted(localResultDict.iteritems(), key=lambda d: int(d[0].replace(".", ""))):
46        print ip + ":\t" + result
47
48if __name__ == "__main__":
49
50    if len(sys.argv) == 2 and sys.argv[1] != "-h":
51        print "======================"
52        print "    Your command is:\t" + sys.argv[1]
53        print "======================"
54        cmd_str = sys.argv[1]
55    else:
56        print """
57            该脚本可以在集群中批量执行任意命令并返回结果，但需注意以下几点：
58            1、命令请先单机测试通过，然后提交给脚本批量执行；
59            2、不要执行 rm 等危险 || 极其耗时 || 影响机器性能的命令；
60            3、命令请用双引号引起来，另外命令中有 $ 符号需要转义成 \$ 否则会被 Shell 当做变量解析掉，具体请参见下面的例子。
61            Usage && for example:
62            python exec_cmd.py "ls -lrt /opt/data1/logs/nginx/pc/track/{}|awk '{{s+=\$5}}END{{print s}}'"
63            python exec_cmd.py "wc -l /opt/data1/logs/nginx/pc/track/{}/*|awk '{{s+=\$1}}END{{print s}}'"
64        """.format(yesterday, yesterday)
65        sys.exit(1)
66
67    exec_cmd(cmd_str)
68
69
70Monitor_RPC_Server.py
71
72#!/usr/bin/env python
73# coding=utf-8
74# 测试utf-8编码
75# cd /opt/script/rpcMonitorFlume
76# pkill -f flumeFileMonitor_RPC_Server.py
77# nohup python -u flumeFileMonitor_RPC_Server.py >> logs/flumeFileMonitor_RPC_Server.log 2>&1 &
78import sys
79reload(sys)
80sys.setdefaultencoding('utf-8')
81
82import os, commands, glob, re
83import datetime
84
85from rpyc import Service
86from rpyc.utils.server import ThreadedServer
87
88from pyUtil import getNowTime, get_ip_address
89
90class remote_call_func(Service):
91
92    def on_connect(self):
93        print "[{0}]\t--------------<<< on_connect".format(getNowTime())
94
95    def on_disconnect(self):
96        print "[{0}]\t-------------->>> on_disconnect".format(getNowTime())
97
98    def exposed_execCmd(self, cmd):
99        exitCode, execResult = commands.getstatusoutput(cmd)
100        nowTime = (datetime.datetime.now()).strftime("%Y-%m-%d %H:%M:%S")
101        print "[{0}] → {1} → {2}".format(nowTime, cmd, execResult)
102        return execResult
103
104
105rpycServer = ThreadedServer(remote_call_func, hostname=get_ip_address('eth0'), port=11111, auto_register=False)
106rpycServer.start()

官方文档中类似例子很多，就不详细介绍了，需注意3点：

• server端定义方法需要被client调用，必须定义以exposed 开头的方法,不然会报错AttributeError: ‘remote_call_script’ object has no attribute ‘exposed_iamshell’

• server端默认不设认证机制，如果需要认证有推荐两种方法: ThreadedServer的authenticator参数与SSL模块

• pip install rpyc ，如果 import rpyc 报错则 yum install openssl-devel，然后重新编译、安装 python

当然还需要考虑很多异常处理，如超时、验证失败等。

Refer：

[1] python远程调用脚本(一)

http://www.dbunix.com/?p=3262

http://rpyc.readthedocs.org/en/latest/tutorial.html

[2] python学习——python中执行shell命令

http://zhou123.blog.51cto.com/4355617/1312791

[3] celery实现任务统一收集、分发执行 

http://blog.csdn.net/vintage_1/article/details/47664187

[4] Timeout function if it takes too long to finish [duplicate]

http://stackoverflow.com/questions/2281850/timeout-function-if-it-takes-too-long-to-finish

[5] 源码之Queue

http://www.cnblogs.com/liqxd/p/5104051.html

[6] python多线程编程(9) Queue模块

http://beginman.cn/python/2015/12/01/python-threading-queue/

[7] Python 并行任务技巧

http://my.oschina.net/leejun2005/blog/194270?fromerr=mNcoWQlp

[8] 利用 Python yield 创建协程将异步编程同步化

http://my.oschina.net/leejun2005/blog/501448?fromerr=ynpLsTXB

[9] Python 多线程教程：并发与并行

http://my.oschina.net/leejun2005/blog/398826

[10] 理解 Python 中的多线程

http://my.oschina.net/leejun2005/blog/179265

[11] paramiko小记

http://www.cnblogs.com/liqxd/p/5077803.html