2019獨角獸企業重金招聘Python工程師標準>>>

原文：https://www.3qos.com/article/101.html

作者：容易 ?日期：2015-03-17

備注：非本人同意，請勿轉載

---

????接上一篇 容易的linux集中管理工具之master端(一)

??? client端代碼

#-*-?coding:?UTF-8?-*-?__author__?=?'tiger'#!/usr/bin/env?pythonimport?zmq,?time,?sys,?os,?atexitfrom?signal?import?SIGTERMimport?randomimport?logging#import?ConfigParserimport?subprocessfrom?logging.handlers?import?RotatingFileHandlerimport?socketimport?fcntlimport?structimport?signal#定義日志函數def?mylog(logfile):rthandler?=?RotatingFileHandler(logfile,?'a',?maxBytes=50?*?1024?*?1024,?backupCount=3)formatter?=?logging.Formatter('%(levelname)s?%(thread)d?%(threadName)s?%(process)d?%(funcName)s?%(asctime)s?%(filename)s[line:%(lineno)d]?%(message)s',datefmt='%a,?%d?%b?%Y?%H:%M:%S')rthandler.setFormatter(formatter)log?=?logging.getLogger()log.setLevel(logging.INFO)log.addHandler(rthandler)return?log#獲取指定接口的IP地址def?get_ip_address(ifname):s?=?socket.socket(socket.AF_INET,?socket.SOCK_DGRAM)return?socket.inet_ntoa(fcntl.ioctl(s.fileno(),0x8915,?#?SIOCGIFADDRstruct.pack('256s',?ifname[:15]))[20:24])#定義命令或者腳本執行超時所觸發的錯誤class?ProcessTimeout(Exception):pass#定義命令或者腳本執行超時所觸發的錯誤句柄def?timeout_handler(signum,?frame):raise?ProcessTimeout#執行命令的函數def?exec_shell(task,?rundir=None,?timeout=None):#定義超時if?timeout:signal.signal(signal.SIGALRM,?timeout_handler)signal.alarm(timeout)p?=?subprocess.Popen(task,?bufsize=0,?shell=True,?cwd=rundir,?stdout=subprocess.PIPE,?stderr=subprocess.PIPE)try:stdout,?stderr?=?p.communicate()returncode?=?p.poll()signal.alarm(0)except?ProcessTimeout:p.stdout.close()p.stderr.close()p.terminate()stderr?=?'Calculation?was?taking?too?long,?so?I?killed?it?dead.\n'returncode?=?'timeout'del?p#返回的信息，如果成功返回標準輸出#錯誤返回標準錯誤if?returncode?==?0:return?[returncode,?stdout]else:return?[returncode,?stderr]#任務類別為腳本時調用該函數def?exec_script(fileserver,?script,?dirpath,?rundir=None,?timeout=None):fileurl?=?fileserver?+?script['script_name']#去http服務器下載腳本，直接調用系統的wget命令下載getfile?=?'wget?-N?-P?'?+?dirpath?+?'?'?+?fileurlfilename?=?dirpath?+?script['script_name']task?=?script['script_env']?+?'?'?+?filenamegetfile_res?=?exec_shell(getfile,?rundir=rundir,?timeout=timeout)if?getfile_res[0]?==?0:task_res?=?exec_shell(task,?rundir=rundir,?timeout=timeout)try:os.remove(filename)except?Exception,?err:task_res[1]?=?task_res[1]?+?str(err)return?task_reselse:return?getfile_res#與master建立連接完成與master的通信def?ioloop(sock_file,?homedir,?mip,?stdout):#定義運行目錄，默認情況下，腳或者命令在該目錄執行dirpath?=?homedir?+?'/run/'#定義日常函數log?=?mylog(stdout)context?=?zmq.Context()socket?=?context.socket(zmq.REQ)socket.setsockopt(zmq.LINGER,?0)socket.connect(sock_file)poller?=?zmq.Poller()poller.register(socket,?zmq.POLLIN)#循環while?1:#發送請求信息try:socket.send_pyobj({'req_type':?'task',?'ip':?mip})except?Exception,?err:log.warn(str(err))socket.close()context.term()time.sleep(random.randint(1,?5))context?=?zmq.Context()socket?=?context.socket(zmq.REQ)socket.setsockopt(zmq.LINGER,?0)socket.connect(sock_file)poller?=?zmq.Poller()poller.register(socket,?zmq.POLLIN)continue#服務端響應超時if?poller.poll(20?*?1000):rep?=?socket.recv_pyobj()#如果有響應信息，判斷響應的類別。try:rep_type?=?rep['rep_type']except?Exception,?err:log.info(str(err))time.sleep(random.uniform(0.8,?1.2))continue#如果響應類別為newtask，則獲取本次任務所需的其他信息if?rep_type?==?'newtask':try:job_id?=?rep['id']job_task?=?rep['task']job_type?=?rep['type']cmd_timeout?=?rep['cmdtimeout']rundir?=?rep['rundir']log.warn('start?new?job?'?+?str(rep))except?Exception,?err:if?job_id:socket.send_pyobj({'id':?job_id,?'code':?'99',?'info':?str(err),?'ip':?mip,?'req_type':?'report'})socket.recv_pyobj()time.sleep(random.uniform(0.8,?1.2))log.warn(str(err)?+?str(rep))continue#如果任務類別是腳本，則嘗試獲取執行腳本所需的其他信息if?job_type?==?'s':try:script_env?=?rep['env']script?=?{'script_name':?job_task,?'script_env':?script_env}fileserver?=?rep['fileserver']#調用運行腳本的函數執行腳本if?rundir?==?'None':res?=?exec_script(fileserver,?script,?dirpath,?rundir=dirpath,?timeout=cmd_timeout)else:res?=?exec_script(fileserver,?script,?dirpath,?rundir=rundir,?timeout=cmd_timeout)except?Exception,?err:log.warn(str(err))continue#任務類別為其他時則統一當作命令執行else:if?rundir?==?'None':res?=?exec_shell(job_task,?rundir=dirpath,?timeout=cmd_timeout)else:res?=?exec_shell(job_task,?rundir=rundir,?timeout=cmd_timeout)#將執行結果返回給master，標記請求類別為report，然master知道該請求是任務報告請求。socket.send_pyobj({'id':?job_id,?'code':?res[0],?'info':?res[1],?'ip':?mip,?'req_type':?'report'})socket.recv_pyobj()log.info(str({'id':?job_id,?'code':?res[0],?'info':?res[1],?'ip':?mip}))time.sleep(random.uniform(0.8,?1.2))else:time.sleep(random.uniform(0.8,?1.2))else:#響應超時時嘗試重連master端log.warn("master?server?connect?time?out,will?colse?current?socket,try?again.")socket.close()context.term()time.sleep(random.randint(1,?5))context?=?zmq.Context()socket?=?context.socket(zmq.REQ)socket.setsockopt(zmq.LINGER,?0)socket.connect(sock_file)poller?=?zmq.Poller()poller.register(socket,?zmq.POLLIN)socket.close()context.term()def?ip_check(ip):q?=?ip.split('.')return?len(q)?==?4?and?len(filter(lambda?x:?x?>=?0?and?x?<=?255,?\map(int,?filter(lambda?x:?x.isdigit(),?q))))?==?4class?Daemon:def?__init__(self,?pidfile,?sock_file,?homedir,?mip,?stdin='/dev/null',?stdout='/dev/null',stderr='/dev/null'):self.stdin?=?stdinself.stdout?=?stdoutself.stderr?=?stderrself.pidfile?=?pidfileself.homedir?=?homedirself.sock_file?=?sock_fileself.mip?=?mipdef?_daemonize(self):#脫離父進程try:pid?=?os.fork()if?pid?>?0:sys.exit(0)except?OSError,?e:sys.stderr.write("fork?#1?failed:?%d?(%s)\n"?%?(e.errno,?e.strerror))sys.exit(1)#脫離終端os.setsid()#修改當前工作目錄os.chdir("/")#重設文件創建權限os.umask(0)#第二次fork，禁止進程重新打開控制終端try:pid?=?os.fork()if?pid?>?0:sys.exit(0)except?OSError,?e:sys.stderr.write("fork?#2?failed:?%d?(%s)\n"?%?(e.errno,?e.strerror))sys.exit(1)sys.stdout.flush()sys.stderr.flush()si?=?file(self.stdin,?'r')so?=?file(self.stdout,?'a+')se?=?file(self.stderr,?'a+',?0)#重定向標準輸入/輸出/錯誤os.dup2(si.fileno(),?sys.stdin.fileno())os.dup2(so.fileno(),?sys.stdout.fileno())os.dup2(se.fileno(),?sys.stderr.fileno())#注冊程序退出時的函數，即刪掉pid文件atexit.register(self.delpid)pid?=?str(os.getpid())file(self.pidfile,?'w+').write("%s\n"?%?pid)def?delpid(self):os.remove(self.pidfile)def?start(self):#?Check?for?a?pidfile?to?see?if?the?daemon?already?runstry:pf?=?file(self.pidfile,?'r')pid?=?int(pf.read().strip())pf.close()except?IOError:pid?=?Noneif?pid:message?=?"pidfile?%s?already?exist.?Daemon?already?running?\n"sys.stderr.write(message?%?self.pidfile)sys.exit(1)#?Start?the?daemonself._daemonize()self._run()def?stop(self):#?Get?the?pid?from?the?pidfiletry:pf?=?file(self.pidfile,?'r')pid?=?int(pf.read().strip())pf.close()except?IOError:pid?=?Noneif?not?pid:message?=?"pidfile?%s?does?not?exist.?Daemon?not?running?\n"sys.stderr.write(message?%?self.pidfile)return?#?not?an?error?in?a?restart#?Try?killing?the?daemon?processtry:while?1:os.kill(pid,?SIGTERM)time.sleep(0.1)except?OSError,?err:err?=?str(err)if?err.find("No?such?process")?>?0:if?os.path.exists(self.pidfile):os.remove(self.pidfile)else:print?str(err)sys.exit(1)def?restart(self):self.stop()self.start()def?_run(self):pass#重寫Daemon的_run函數實現自己的Daemon進程class?MyDaemon(Daemon):def?_run(self,?):ioloop(self.sock_file,?self.homedir,?self.mip,?self.stdout)#定義主函數，創建相關運行目錄和定義日志路徑等def?main():homedir?=?os.getcwd()for?i?in?('log',?'run'):path?=?homedir?+?'/'?+?iif?not?os.path.exists(path):os.makedirs(path,?0755)stdout?=?homedir?+?'/log'?+?'/oaos_client.log'stderr?=?homedir?+?'/log'?+?'/oaos_client.err'pidfile?=?homedir?+?'/run'?+?'/oaos_client.pid'#master的tcp接口sock_file?=?"tcp://192.168.4.194:7777"#該接口是指用來與master通信的客戶端IP接口ifname?=?'eth0'try:mip?=?get_ip_address(ifname)except?Exception,?err:print?errsys.exit(3)daemon?=?MyDaemon(pidfile,?sock_file,?homedir,?mip,?stdout=stdout,?stderr=stderr)if?len(sys.argv)?==?2:if?'start'?==?sys.argv[1]:daemon.start()elif?'stop'?==?sys.argv[1]:daemon.stop()elif?'restart'?==?sys.argv[1]:daemon.restart()else:print?"Unknown?command"sys.exit(2)sys.exit(0)else:print?"usage:?%s?start|stop|restart"?%?sys.argv[0]sys.exit(2)if?__name__?==?"__main__":main()

轉載于:https://my.oschina.net/u/245398/blog/389976

總結

以上是生活随笔為你收集整理的容易的linux自动化运维工具之clinet端(二)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。