進程管理工具Supervisor(二)Events
supervisor可以當做一個簡單的進程啟動、重啟、控制工具使用,也可以作為一個進程監控框架使用,作為後者,需要使用supervisor的Events機制。
Event Listeners
supervisor對子程序的監控通過叫做event listener的程序實現。supervisor控制的子程序狀態發生變化時,就會產生一些事件通知,event listener可以對這些事件通知進行訂閱。
event listener本身也是作為supervisor的子程序運行的。事件通知協議的實現基於event listener子程序的stdin和stdout。supervisor發送特定格式的信息到event listener的stdin,然後從event listener的stdout獲得特定格式的輸出,從而形成一個請求/應答循環。
配置
event listener的配置放置於配置文件中的[eventlistener:x]塊中。
[eventlistener:mylistener] command=my_custom_listener.py events=PROCESS_STATE,TICK_60
x是listener的名稱,command是執行listener腳本的命令,events是要監控的事件類型。
event listener本身是作為supervisor的子程序運行的,所以與配置子程序[program:x]塊類似,官網例子:
[eventlistener:theeventlistenername] command=/bin/eventlistener process_name=%(program_name)s_%(process_num)02d numprocs=5 events=PROCESS_STATE buffer_size=10 directory=/tmp umask=022 priority=-1 autostart=true autorestart=unexpected startsecs=1 startretries=3 exitcodes=0,2 stopsignal=QUIT stopwaitsecs=10 stopasgroup=false killasgroup=false user=chrism redirect_stderr=false stdout_logfile=/a/path stdout_logfile_maxbytes=1MB stdout_logfile_backups=10 stdout_events_enabled=false stderr_logfile=/a/path stderr_logfile_maxbytes=1MB stderr_logfile_backups=10 stderr_events_enabled=false environment=A="1",B="2" serverurl=AUTO
事件通知協議
一個event listener可以處於三種狀態,ACKNOWLEDGED、READY、BUSY,只有在READY狀態下才可以接收事件通知。
event listener啟動時處於ACKNOWLEDGED狀態,直到event listener向stdout中輸出“READY\n”字符串為止。
event listener向stdout中輸出“READY\n”之後就處於READY狀態,supervisor會向處於READY狀態的listener發送listener訂閱的事件通知。
listener接收事件通知之後就處於BUSY狀態,期間listener對接收到的事件通知進行處理,處理結束後向stdout輸出“
supervisor收到OK或者FAIL輸出後,就將event listener的狀態置於ACKNOWLEDGED。FAIL的事件通知會被緩存然後再次發送。
event listener的狀態處於ACKNOWLEDGED後可以退出執行,也可以繼續執行,繼續執行就可以向stdout輸出“READY\n”形成一個循環。
supervisor向listener發送的事件通知由兩部分組成,header和body,由"\n"換行符分開。
一個header例子:
ver:3.0 server:supervisor serial:21 pool:listener poolserial:10 eventname:PROCESS_COMMUNICATION_STDOUT len:54
ver:協議版本
server:supervisor的標識符,由[supervisord]塊中的identifier選項設置。
serial:event的序列號
pool:listener的pool的名字。
poolserial:event在pool中的的序列號
eventname:event類型名稱
len:header後面的body長度。
一個body例子:
processname:foo groupname:bar pid:123 This is the data that was sent between the tags
processname:事件所屬的子進程名字
groupname:子進程所屬組名
pid:子進程pid
一個簡單的listener腳本,listener.py:
import sys def write_stdout(s): # only eventlistener protocol messages may be sent to stdout sys.stdout.write(s) sys.stdout.flush() def write_stderr(s): sys.stderr.write(s) sys.stderr.flush() def main(): while True: # 進入READY狀態 ┆ write_stdout(‘READY\n‘) ┆ # 讀取事件通知的header ┆ line = sys.stdin.readline() ┆ write_stderr(line) # 獲取body長度,讀取body ┆ headers=dict([x.split(‘:‘) for x in line.split() ]) ┆ data = sys.stdin.read(int(headers[‘len‘])) ┆ write_stderr(data+‘\n‘) # 發送OK進入ACKNOWLEDGED狀態 ┆ write_stdout(‘RESULT 2\nOK‘) if __name__ == ‘__main__‘: main()
在conf.d目錄中建立一個listener配置文件mylistener.conf:
[eventlistener:mylistener] command=python listener.py directory=/thedirectoroflistener.py user=user events=PROCESS_STATE,TICK_5 stdout_logfile=/path/to/mylistener_stdout.log stderr_logfile=/path/to/mylistener_stderr.log
啟動:
ubuntu:$ sudo supervisorctl start all mylistener: started celerybeat: started ubuntu:$ sudo supervisorctl status celerybeat RUNNING pid 87729, uptime 0:00:20 mylistener RUNNING pid 87728, uptime 0:00:20
監控就開始了,可以到日誌中查看事件通知的內容:
ver:3.0 server:supervisor serial:15361 pool:mylistener poolserial:15361 eventname:PROCESS_STATE_RUNNING len:73 processname:mylistener groupname:mylistener from_state:STARTING pid:87728 ver:3.0 server:supervisor serial:15362 pool:mylistener poolserial:15362 eventname:TICK_5 len:15 when:1514313560 ver:3.0 server:supervisor serial:15364 pool:mylistener poolserial:15364 eventname:PROCESS_STATE_RUNNING len:73 processname:celerybeat groupname:celerybeat from_state:STARTING pid:87729
可以根據自己的需要設定監控的事件類型,然後根據不同的事件類型和內容做出不同的應變,具體的事件類型可以官網查看。
python的supervisor.childutils模塊對header和body的處理進行了包裝:
def get_headers(line): return dict([ x.split(‘:‘) for x in line.split() ]) def eventdata(payload): headerinfo, data = payload.split(‘\n‘, 1) headers = get_headers(headerinfo) return headers, data def get_asctime(now=None): if now is None: # for testing now = time.time() # pragma: no cover msecs = (now - long(now)) * 1000 part1 = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now)) asctime = ‘%s,%03d‘ % (part1, msecs) return asctime class ProcessCommunicationsProtocol: def send(self, msg, fp=sys.stdout): fp.write(ProcessCommunicationEvent.BEGIN_TOKEN) fp.write(msg) fp.write(ProcessCommunicationEvent.END_TOKEN) def stdout(self, msg): return self.send(msg, sys.stdout) def stderr(self, msg): return self.send(msg, sys.stderr) pcomm = ProcessCommunicationsProtocol() class EventListenerProtocol: def wait(self, stdin=sys.stdin, stdout=sys.stdout): self.ready(stdout) line = stdin.readline() headers = get_headers(line) payload = stdin.read(int(headers[‘len‘])) return headers, payload def ready(self, stdout=sys.stdout): stdout.write(PEventListenerDispatcher.READY_FOR_EVENTS_TOKEN) stdout.flush() def ok(self, stdout=sys.stdout): self.send(‘OK‘, stdout) def fail(self, stdout=sys.stdout): self.send(‘FAIL‘, stdout) def send(self, data, stdout=sys.stdout): resultlen = len(data) result = ‘%s%s\n%s‘ % (PEventListenerDispatcher.RESULT_TOKEN_START, str(resultlen), data) stdout.write(result) stdout.flush() listener = EventListenerProtocol()
listener腳本可以方便的寫做:
import sys from supervisor import childutils def write_stdout(s): # only eventlistener protocol messages may be sent to stdout sys.stdout.write(s) sys.stdout.flush() def write_stderr(s): sys.stderr.write(s) sys.stderr.flush() def main(): while True: ┆ headers, payload = childutils.listener.wait() ┆ write_stderr(payload+‘\n‘) ┆ childutils.listener.ok(sys.stdout) if __name__ == ‘__main__‘: main()
進程管理工具Supervisor(二)Events