上篇文章介紹了supervisor的使用， 今天介紹一下supervisor的源碼。 supervisor是python寫的。如果你不懂python，我也介紹一下golang版的實現(xiàn)。 如果你不會，其他語言的實現(xiàn)，應該也有，github找一下。 自己讀一下。實現(xiàn)原理應該都是一樣的。
supervisor的組件
1. supervisord
服務器主進程名為supervisord。它負責在自己的調(diào)用中啟動子程序，響應來自客戶端的命令，重新啟動崩潰或退出的子進程，記錄其子進程stdout和stderr 輸出，以及生成和處理與子進程生命周期中的點相對應的“事件”。
服務器進程使用配置文件。這通常位于/etc/supervisord.conf中。此配置文件是“Windows-INI”樣式配置文件。通過適當?shù)奈募到y(tǒng)權限保持此文件的安全非常重要，因為它可能包含未加密的用戶名和密碼。
2. supervisorctl
主進程的命令行客戶端部分名為 supervisorctl。它為supervisord提供的功能提供了類似shell的界面。從 supervisorctl，用戶可以連接到不同的 supervisord進程（一次一個），獲取由子進程控制的狀態(tài)，停止和啟動子進程，并獲取supervisord的運行進程列表。
命令行客戶機通過UNIX域套接字或internet (TCP)套接字與服務器通信。服務器可以斷言客戶機的用戶應該在執(zhí)行命令之前提供身份驗證憑據(jù)。客戶機進程通常使用與服務器相同的配置文件，但是其中包含[supervisorctl]部分的任何配置文件都可以工作。
3. Web Server
如果您針對internet套接字啟動了adminord，那么可以通過瀏覽器訪問具有與supervise orctl類似功能的(稀疏的)web用戶界面。在激活配置文件的[inet_http_server]部分之后，訪問服務器URL(例如http://localhost:9001/)，通過web界面查看和控制進程狀態(tài)。
4. XML-RPC Interface
服務于web UI的同一個HTTP服務器提供一個XML-RPC接口，該接口可用于詢問和控制管理器及其運行的程序。參見XML-RPC API文檔。
整體看一下目錄結構：
├── __init__.py
├── childutils.py
├── compat.py
├── confecho.py
├── datatypes.py
├── dispatchers.py
├── events.py
├── http.py
├── http_client.py
├── loggers.py
├── medusa
│ ├── CHANGES.txt
│ ├── LICENSE.txt
│ ├── README.txt
│ ├── TODO.txt
│ ├── __init__.py
│ ├── asynchat_25.py
│ ├── asyncore_25.py
│ ├── auth_handler.py
│ ├── counter.py
│ ├── default_handler.py
│ ├── docs
│ │ ├── README.html
│ │ ├── async_blurbs.txt
│ │ ├── composing_producers.gif
│ │ ├── data_flow.gif
│ │ ├── data_flow.html
│ │ ├── producers.gif
│ │ ├── programming.html
│ │ ├── proxy_notes.txt
│ │ ├── threads.txt
│ │ └── tkinter.txt
│ ├── filesys.py
│ ├── http_date.py
│ ├── http_server.py
│ ├── logger.py
│ ├── producers.py
│ ├── util.py
│ └── xmlrpc_handler.py
├── options.py
├── pidproxy.py
├── poller.py
├── process.py
├── rpcinterface.py
├── scripts
│ ├── loop_eventgen.py
│ ├── loop_listener.py
│ ├── sample_commevent.py
│ ├── sample_eventlistener.py
│ └── sample_exiting_eventlistener.py
├── skel
│ └── sample.conf
├── socket_manager.py
├── states.py
├── supervisorctl.py
├── supervisord.py
├── tests
│ ├── __init__.py
│ ├── base.py
│ ├── fixtures
│ │ ├── donothing.conf
│ │ ├── example
│ │ │ └── included.conf
│ │ ├── hello.sh
│ │ ├── include.conf
│ │ ├── issue-1054.conf
│ │ ├── issue-565.conf
│ │ ├── issue-638.conf
│ │ ├── issue-663.conf
│ │ ├── issue-664.conf
│ │ ├── issue-835.conf
│ │ ├── issue-836.conf
│ │ ├── listener.py
│ │ ├── spew.py
│ │ └── unkillable_spew.py
│ ├── test_childutils.py
│ ├── test_confecho.py
│ ├── test_datatypes.py
│ ├── test_dispatchers.py
│ ├── test_end_to_end.py
│ ├── test_events.py
│ ├── test_http.py
│ ├── test_http_client.py
│ ├── test_loggers.py
│ ├── test_options.py
│ ├── test_poller.py
│ ├── test_process.py
│ ├── test_rpcinterfaces.py
│ ├── test_socket_manager.py
│ ├── test_states.py
│ ├── test_supervisorctl.py
│ ├── test_supervisord.py
│ ├── test_web.py
│ └── test_xmlrpc.py
├── ui
│ ├── images
│ │ ├── button_refresh.gif
│ │ ├── button_restart.gif
│ │ ├── button_stop.gif
│ │ ├── icon.png
│ │ ├── rule.gif
│ │ ├── state0.gif
│ │ ├── state1.gif
│ │ ├── state2.gif
│ │ ├── state3.gif
│ │ └── supervisor.gif
│ ├── status.html
│ ├── stylesheets
│ │ └── supervisor.css
│ └── tail.html
├── version.txt
├── web.py
└── xmlrpc.py
整體架構
源碼學習按上面的分塊進行介紹
supervisord
先找到入口程序
# Main program
def main(args=None, test=False):
assert os.name == "posix", "This code makes Unix-specific assumptions"
# if we hup, restart by making a new Supervisor()
first = True
while 1:
options = ServerOptions()
options.realize(args, doc=__doc__)
options.first = first
options.test = test
if options.profile_options:
sort_order, callers = options.profile_options
profile('go(options)', globals(), locals(), sort_order, callers)
else:
go(options)
options.close_httpservers()
options.close_logger()
first = False
if test or (options.mood < SupervisorStates.RESTARTING):
break
if __name__ == "__main__": # pragma: no cover
main()
從上面看出， main()中有個死循環(huán)一直只工作。
下面詳細介紹，在循環(huán)中做了哪些工作？
# Main program
def main(args=None, test=False):
assert os.name == "posix", "This code makes Unix-specific assumptions"
# if we hup, restart by making a new Supervisor()
first = True
while 1:
options = ServerOptions() // 配置
options.realize(args, doc=__doc__)
options.first = first
options.test = test
if options.profile_options:
sort_order, callers = options.profile_options
profile('go(options)', globals(), locals(), sort_order, callers)
else:
go(options) // 加載配置開始運行
options.close_httpservers()
options.close_logger()
first = False
if test or (options.mood < SupervisorStates.RESTARTING):
break
def go(options): # pragma: no cover
d = Supervisor(options) // 實例化一個Supervisor對象
try:
d.main() // 運行main()函數(shù)
except asyncore.ExitNow:
pass
Supervisor類的代碼
class Supervisor:
stopping = False # set after we detect that we are handling a stop request
lastshutdownreport = 0 # throttle for delayed process error reports at stop
process_groups = None # map of process group name to process group object
stop_groups = None # list used for priority ordered shutdown
def __init__(self, options): # 初始化
self.options = options # 配置
self.process_groups = {}
self.ticks = {}
def main(self):
if not self.options.first:
# prevent crash on libdispatch-based systems, at least for the
# first request
self.options.cleanup_fds()
self.options.set_uid_or_exit()
if self.options.first:
self.options.set_rlimits_or_exit()
# this sets the options.logger object
# delay logger instantiation until after setuid
self.options.make_logger()
if not self.options.nocleanup:
# clean up old automatic logs
self.options.clear_autochildlogdir()
self.run() # 運行
def run(self):
self.process_groups = {} # clear
self.stop_groups = None # clear
events.clear()
try:
# 根據(jù)配置進行添加process
for config in self.options.process_group_configs:
self.add_process_group(config)
# 進程環(huán)境
self.options.process_environment()
# 打開http web
self.options.openhttpservers(self)
# 用于捕獲信號
self.options.setsignals()
# 主進程是否成為守護進程
if (not self.options.nodaemon) and self.options.first:
self.options.daemonize()
# writing pid file needs to come *after* daemonizing or pid
# will be wrong
self.options.write_pidfile()
# 運行異步io服務器
self.runforever()
finally:
# 異常退出，清理工作
self.options.cleanup()
上面代碼只有 self.runforever() 是工作的
def runforever(self):
# 事件通知機制
events.notify(events.SupervisorRunningEvent())
timeout = 1 # this cannot be fewer than the smallest TickEvent (5)
# 獲取已經(jīng)注冊的句柄
socket_map = self.options.get_socket_map()
# 這里會一直 運行，相當于守護進程
while 1:
# 保存運行信息等
combined_map = {}
combined_map.update(socket_map)
combined_map.update(self.get_process_map())
# 進程信息
pgroups = list(self.process_groups.values())
pgroups.sort()
# 根據(jù)進程配置開啟或關閉進程
if self.options.mood < SupervisorStates.RUNNING:
if not self.stopping:
# first time, set the stopping flag, do a
# notification and set stop_groups
self.stopping = True
self.stop_groups = pgroups[:]
events.notify(events.SupervisorStoppingEvent())
self.ordered_stop_groups_phase_1()
if not self.shutdown_report():
# if there are no unstopped processes (we're done
# killing everything), it's OK to shutdown or reload
raise asyncore.ExitNow
for fd, dispatcher in combined_map.items():
if dispatcher.readable():
self.options.poller.register_readable(fd)
if dispatcher.writable():
self.options.poller.register_writable(fd)
# poll操作
r, w = self.options.poller.poll(timeout)
for fd in r:
if fd in combined_map:
try:
dispatcher = combined_map[fd]
self.options.logger.blather(
'read event caused by %(dispatcher)r',
dispatcher=dispatcher)
dispatcher.handle_read_event()
if not dispatcher.readable():
self.options.poller.unregister_readable(fd)
except asyncore.ExitNow:
raise
except:
combined_map[fd].handle_error()
# 依次遍歷注冊的文件句柄
for fd in w:
if fd in combined_map:
try:
dispatcher = combined_map[fd]
self.options.logger.blather(
'write event caused by %(dispatcher)r',
dispatcher=dispatcher)
dispatcher.handle_write_event()
if not dispatcher.writable():
self.options.poller.unregister_writable(fd)
except asyncore.ExitNow:
raise
except:
combined_map[fd].handle_error()
for group in pgroups:
group.transition()
# 獲取已經(jīng)死亡的子進程信息
self.reap()
# 處理信號
self.handle_signal()
# tick時鐘
self.tick()
if self.options.mood < SupervisorStates.RUNNING:
self.ordered_stop_groups_phase_2()
if self.options.test:
break
下面詳細介紹一下如何管理一個process：
下面圖可以指導主要是
def add_process_group(self, config):
name = config.name
if name not in self.process_groups:
config.after_setuid()
# 根據(jù)初始化后的配置文件生成相應的子進程實例
self.process_groups[name] = config.make_group()
# 添加事件通知
events.notify(events.ProcessGroupAddedEvent(name))
return True
return False
在supervisor配置中，我們需要寫上執(zhí)行的文件以及執(zhí)行環(huán)境env. 所以，我們圍繞 執(zhí)行這個程序的邏輯就可以了
def execve(self, filename, argv, env):
return os.execve(filename, argv, env)
作者將所有的process或event的配置 都綁定到 options的對象上， 包括執(zhí)行的程序以及各種狀態(tài)， 然后一個deamon程序一直在運行，去實時檢查配置是否發(fā)生變化了。做對應的操作。
supervisorclt
這個客戶端功能：
- 需要有一個UI,實現(xiàn)是cli
- 與server通信
def main(args=None, options=None):
if options is None:
# 實例化對象
options = ClientOptions()
options.realize(args, doc=__doc__)
# 控制器
c = Controller(options)
if options.args:
c.onecmd(" ".join(options.args))
sys.exit(c.exitstatus)
# 如果是交互模式
if options.interactive:
# 一直loop, 圖形編程常見操作模式
c.exec_cmdloop(args, options)
sys.exit(0) # exitstatus always 0 for interactive mode
我們看交互式的邏輯，exec_cmdloop
# 控制器的method
def exec_cmdloop(self, args, options):
try:
import readline
delims = readline.get_completer_delims()
delims = delims.replace(':', '') # "group:process" as one word
delims = delims.replace('*', '') # "group:*" as one word
delims = delims.replace('-', '') # names with "-" as one word
readline.set_completer_delims(delims)
if options.history_file:
try:
readline.read_history_file(options.history_file)
except IOError:
pass
def save():
try:
readline.write_history_file(options.history_file)
except IOError:
pass
import atexit
atexit.register(save)
except ImportError:
pass
try:
self.cmdqueue.append('status')
# cmdloop
self.cmdloop()
except KeyboardInterrupt:
self.output('')
pass
# cmdloop method定義
def cmdloop(self, intro=None):
"""Repeatedly issue a prompt, accept input, parse an initial prefix
off the received input, and dispatch to action methods, passing them
the remainder of the line as argument.
"""
self.preloop()
# 輸入和快捷鍵
if self.use_rawinput and self.completekey:
try:
import readline
self.old_completer = readline.get_completer()
readline.set_completer(self.complete)
readline.parse_and_bind(self.completekey+": complete")
except ImportError:
pass
try:
if intro is not None:
self.intro = intro
if self.intro:
self.stdout.write(str(self.intro)+"\n")
stop = None
while not stop:
if self.cmdqueue:
line = self.cmdqueue.pop(0)
else:
if self.use_rawinput:
try:
line = input(self.prompt)
except EOFError:
line = 'EOF'
else:
self.stdout.write(self.prompt)
self.stdout.flush()
line = self.stdin.readline()
if not len(line):
line = 'EOF'
else:
line = line.rstrip('\r\n')
line = self.precmd(line)
stop = self.onecmd(line)
stop = self.postcmd(stop, line)
self.postloop()
finally:
if self.use_rawinput and self.completekey:
try:
import readline
readline.set_completer(self.old_completer)
except ImportError:
pass
上面執(zhí)行又會執(zhí)行到onecmd
def onecmd(self, line):
"""Interpret the argument as though it had been typed in response
to the prompt.
This may be overridden, but should not normally need to be;
see the precmd() and postcmd() methods for useful execution hooks.
The return value is a flag indicating whether interpretation of
commands by the interpreter should stop.
"""
cmd, arg, line = self.parseline(line)
if not line:
return self.emptyline()
if cmd is None:
return self.default(line)
self.lastcmd = line
if line == 'EOF' :
self.lastcmd = ''
if cmd == '':
return self.default(line)
else:
try:
func = getattr(self, 'do_' + cmd)
except AttributeError:
return self.default(line)
return func(arg)
可以看出 會返回一個func = getattr(self, 'do_' + cmd),函數(shù)調(diào)用，是以do_開頭的函數(shù)
在supervisorctl的源碼文件中， 我們找一個示例
def do_start(self, arg):
if not self.ctl.upcheck():
return
names = arg.split()
# 連接supervisor服務的代理
supervisor = self.ctl.get_supervisor()
if not names:
self.ctl.output("Error: start requires a process name")
self.ctl.exitstatus = LSBInitExitStatuses.INVALID_ARGS
self.help_start()
return
# 開啟所有
if 'all' in names:
# 調(diào)用startAllProcesses函數(shù)
results = supervisor.startAllProcesses()
for result in results:
self.ctl.output(self._startresult(result))
self.ctl.set_exitstatus_from_xmlrpc_fault(result['status'], xmlrpc.Faults.ALREADY_STARTED)
else:
# 逐個關閉
for name in names:
group_name, process_name = split_namespec(name)
if process_name is None:
try:
results = supervisor.startProcessGroup(group_name)
for result in results:
self.ctl.output(self._startresult(result))
self.ctl.set_exitstatus_from_xmlrpc_fault(result['status'], xmlrpc.Faults.ALREADY_STARTED)
except xmlrpclib.Fault as e:
if e.faultCode == xmlrpc.Faults.BAD_NAME:
error = "%s: ERROR (no such group)" % group_name
self.ctl.output(error)
self.ctl.exitstatus = LSBInitExitStatuses.INVALID_ARGS
else:
self.ctl.exitstatus = LSBInitExitStatuses.GENERIC
raise
else:
try:
result = supervisor.startProcess(name)
except xmlrpclib.Fault as e:
error = {'status': e.faultCode,
'name': process_name,
'group': group_name,
'description': e.faultString}
self.ctl.output(self._startresult(error))
self.ctl.set_exitstatus_from_xmlrpc_fault(error['status'], xmlrpc.Faults.ALREADY_STARTED)
else:
name = make_namespec(group_name, process_name)
self.ctl.output('%s: started' % name)
上面比較重要的就是遠程通過rpc調(diào)用
class SupervisorNamespaceRPCInterface:
....
def startProcessGroup(self, name, wait=True):
""" Start all processes in the group named 'name'
@param string name The group name
@param boolean wait Wait for each process to be fully started
@return array result An array of process status info structs
"""
self._update('startProcessGroup')
group = self.supervisord.process_groups.get(name)
if group is None:
raise RPCError(Faults.BAD_NAME, name)
processes = list(group.processes.values())
processes.sort()
processes = [ (group, process) for process in processes ]
startall = make_allfunc(processes, isNotRunning, self.startProcess,
wait=wait)
startall.delay = 0.05
startall.rpcinterface = self
return startall # deferred
web UI
def make_http_servers(options, supervisord):
from supervisor.web import supervisor_ui_handler
uihandler = supervisor_ui_handler(supervisord)
class supervisor_ui_handler:
IDENT = 'Supervisor Web UI HTTP Request Handler'
def __init__(self, supervisord):
self.supervisord = supervisord
def match(self, request):
if request.command not in ('POST', 'GET'):
return False
path, params, query, fragment = request.split_uri()
while path.startswith('/'):
path = path[1:]
if not path:
path = 'index.html'
for viewname in VIEWS.keys():
if viewname == path:
return True
def handle_request(self, request):
if request.command == 'POST':
request.collector = collector(self, request)
else:
self.continue_request('', request)
def continue_request (self, data, request):
form = {}
cgi_env = request.cgi_environment()
form.update(cgi_env)
if 'QUERY_STRING' not in form:
form['QUERY_STRING'] = ''
query = form['QUERY_STRING']
# we only handle x-www-form-urlencoded values from POSTs
form_urlencoded = parse_qsl(data)
query_data = parse_qs(query)
for k, v in query_data.items():
# ignore dupes
form[k] = v[0]
for k, v in form_urlencoded:
# ignore dupes
form[k] = v
form['SERVER_URL'] = request.get_server_url()
path = form['PATH_INFO']
# strip off all leading slashes
while path and path[0] == '/':
path = path[1:]
if not path:
path = 'index.html'
viewinfo = VIEWS.get(path)
if viewinfo is None:
# this should never happen if our match method works
return
response = {'headers': {}}
viewclass = viewinfo['view']
viewtemplate = viewinfo['template']
context = ViewContext(template=viewtemplate,
request = request,
form = form,
response = response,
supervisord=self.supervisord)
view = viewclass(context)
pushproducer = request.channel.push_with_producer
pushproducer(DeferredWebProducer(request, view))
RPC
class supervisor_xmlrpc_handler(xmlrpc_handler):
path = '/RPC2'
IDENT = 'Supervisor XML-RPC Handler'
unmarshallers = {
"int": lambda x: int(x.text),
"i4": lambda x: int(x.text),
"boolean": lambda x: x.text == "1",
"string": lambda x: x.text or "",
"double": lambda x: float(x.text),
"dateTime.iso8601": lambda x: make_datetime(x.text),
"array": lambda x: x[0].text,
"data": lambda x: [v.text for v in x],
"struct": lambda x: dict([(k.text or "", v.text) for k, v in x]),
"base64": lambda x: as_string(decodestring(as_bytes(x.text or ""))),
"param": lambda x: x[0].text,
}
def __init__(self, supervisord, subinterfaces):
self.rpcinterface = RootRPCInterface(subinterfaces)
self.supervisord = supervisord
def loads(self, data):
params = method = None
for action, elem in iterparse(StringIO(data)):
unmarshall = self.unmarshallers.get(elem.tag)
if unmarshall:
data = unmarshall(elem)
elem.clear()
elem.text = data
elif elem.tag == "value":
try:
data = elem[0].text
except IndexError:
data = elem.text or ""
elem.clear()
elem.text = data
elif elem.tag == "methodName":
method = elem.text
elif elem.tag == "params":
params = tuple([v.text for v in elem])
return params, method
def match(self, request):
return request.uri.startswith(self.path)
def continue_request(self, data, request):
logger = self.supervisord.options.logger
try:
try:
# on 2.x, the Expat parser doesn't like Unicode which actually
# contains non-ASCII characters. It's a bit of a kludge to
# do it conditionally here, but it's down to how underlying
# libs behave
if PY2:
data = data.encode('ascii', 'xmlcharrefreplace')
params, method = self.loads(data)
except:
logger.error(
'XML-RPC request data %r is invalid: unmarshallable' %
(data,)
)
request.error(400)
return
# no <methodName> in the request or name is an empty string
if not method:
logger.error(
'XML-RPC request data %r is invalid: no method name' %
(data,)
)
request.error(400)
return
# we allow xml-rpc clients that do not send empty <params>
# when there are no parameters for the method call
if params is None:
params = ()
try:
logger.trace('XML-RPC method called: %s()' % method)
value = self.call(method, params)
logger.trace('XML-RPC method %s() returned successfully' %
method)
except RPCError as err:
# turn RPCError reported by method into a Fault instance
value = xmlrpclib.Fault(err.code, err.text)
logger.trace('XML-RPC method %s() returned fault: [%d] %s' % (
method,
err.code, err.text))
if isinstance(value, types.FunctionType):
# returning a function from an RPC method implies that
# this needs to be a deferred response (it needs to block).
pushproducer = request.channel.push_with_producer
pushproducer(DeferredXMLRPCResponse(request, value))
else:
# if we get anything but a function, it implies that this
# response doesn't need to be deferred, we can service it
# right away.
body = as_bytes(xmlrpc_marshal(value))
request['Content-Type'] = 'text/xml'
request['Content-Length'] = len(body)
request.push(body)
request.done()
except:
tb = traceback.format_exc()
logger.critical(
"Handling XML-RPC request with data %r raised an unexpected "
"exception: %s" % (data, tb)
)
# internal error, report as HTTP server error
request.error(500)
def call(self, method, params):
return traverse(self.rpcinterface, method, params)
總結
python是解釋型語言 弱類型，看源碼沒有編譯型語言清晰明了。這個源碼不太容易看，我參考supervisor的設計寫公司項目，一開始是看這個源碼，發(fā)現(xiàn)很不容易理清楚，我是先看了go語言版的，然后在回頭看python版的才看明白。我后面整理go版本的，那個更加清晰。
