從今天開始將陸陸續(xù)續(xù)發(fā)表一些openstack相關(guān)的文章措嵌。

openstack服務(wù)的啟動(dòng)

基本所有的openstack服務(wù)都依賴 evenlet 完成各種并發(fā)任務(wù)，它的進(jìn)程可分為兩類：
1蔬捷、?WSGIService: 接收和處理 http 請(qǐng)求，依賴eventlet.wsgi?的?wsgi server?處理 http 請(qǐng)求榔袋，比如nova-api
2周拐、?Service: 接收和處理 rpc 請(qǐng)求，如?nova-operation等
無論是?WSGIService?還是?Service?類型的進(jìn)程凰兑，每當(dāng)接收到一個(gè)請(qǐng)求(http 或 rpc)妥粟，都會(huì)在線程池中分配一個(gè)協(xié)程處理該請(qǐng)求

一、WSGIService的啟動(dòng)

下面以nova服務(wù)為例吏够。
nova-api 由?nova/cmd/api.py?啟動(dòng)勾给，它初始化一個(gè) WSGIService(由?service.py?定義) 對(duì)象滩报。

def main():
    objects.register_all()
    CONF(sys.argv[1:], project='nova',
         version=version.version_string())
    logging.setup(CONF, "nova")

    rpc.init(CONF)
    launcher = service.get_launcher()
    server = service.WSGIService('osapi_nova')
    launcher.launch_service(server, workers=server.workers)
    launcher.wait()

api中從service層獲取一個(gè)啟動(dòng)器對(duì)象，最后將server對(duì)象傳入啟動(dòng)器對(duì)象的launch_service方法中锦秒，launch_service（server, workers=server.workers）方法定義如下：

class Launcher(object):
    def __init__(self):
        super(Launcher, self).__init__()
        self.launch_service = serve
        self.wait = wait

該方法被引用到serve方法露泊，serve方法定義如下：

def serve(server, workers=None):
    global _launcher
    if _launcher:
        raise RuntimeError(_('serve() can only be called once'))

    _launcher = service.launch(CONF, server, workers=workers)

最終調(diào)用了oslo_service/service.py下的launch方法，launch方法定義如下：

def launch(conf, service, workers=1, restart_method='reload'):
    …

    if workers is not None and workers <= 0:
        raise ValueError(_("Number of workers should be positive!"))

    if workers is None or workers == 1:
        launcher = ServiceLauncher(conf, restart_method=restart_method)
    else:
        launcher = ProcessLauncher(conf, restart_method=restart_method)
    launcher.launch_service(service, workers=workers)

可以看到這里使用到了兩種啟動(dòng)器旅择，在進(jìn)一步講解啟動(dòng)的過程中先介紹下openstack中的啟動(dòng)器

二惭笑、Openstack中的Launcher

Openstack中有一個(gè)叫Launcher的概念，即專門用來啟動(dòng)服務(wù)的生真，這個(gè)類被放在了oslo_service這個(gè)包里面沉噩，Launcher分為兩種:
一種是ServiceLauncher；
另一種為ProcessLauncher柱蟀。
ServiceLauncher用來啟動(dòng)單進(jìn)程的服務(wù)川蒙；而ProcessLauncher用來啟動(dòng)有多個(gè)worker子進(jìn)程的服務(wù)，如各類api服務(wù)(nova-api长已、cinder-api)等

oslo_service/service.py

1畜眨、ServiceLauncher

ServiceLauncher繼承自Launcher，啟動(dòng)服務(wù)的一個(gè)重要成員就是launcher_service术瓮，ServiceLauncher的該成員就是繼承于Launcher

def launch_service(self, service, workers=1):
 	…
    if workers is not None and workers != 1:
        raise ValueError(_("Launcher asked to start multiple workers"))
    _check_service_base(service)
    service.backdoor_port = self.backdoor_port
    self.services.add(service)

aucher_service就是將服務(wù)添加到self.services成員里面康聂，services成員的類型是class Services，看看它的add方法

class Services(object):

    def __init__(self):
        self.services = []
        self.tg = threadgroup.ThreadGroup()
        self.done = event.Event()

    def add(self, service):
        """Add a service to a list and create a thread to run it.

        :param service: service to run
        """
        self.services.append(service)
        self.tg.add_thread(self.run_service, service, self.done)

Services這個(gè)類的初始化很簡(jiǎn)單胞四，即創(chuàng)建一個(gè)ThreadGroup恬汁，ThreadGroup其實(shí)是eventlet的GreenPool，Openstack利用eventlet實(shí)現(xiàn)并發(fā)辜伟，add方法氓侧，將self.run_service這個(gè)方法放入pool中，而service就是它的參數(shù)导狡。run_service方法很簡(jiǎn)單约巷，就是調(diào)用service的start方法，這樣就完成了服務(wù)的啟動(dòng)

2旱捧、ProcessLauncher

ProcessLauncher直接繼承于Object载庭，同樣也有l(wèi)aunch_service方法

def launch_service(self, service, workers=1):
 	…
    _check_service_base(service)
    wrap = ServiceWrapper(service, workers)

    LOG.info('Starting %d workers', wrap.workers)
    while self.running and len(wrap.children) < wrap.workers:
        self._start_child(wrap)

lauch_service除了接受service以外，還需要接受一個(gè)workers參數(shù)廊佩，即子進(jìn)程的個(gè)數(shù)，然后調(diào)用_start_child啟動(dòng)多個(gè)子進(jìn)程

def _start_child(self, wrap):
    if len(wrap.forktimes) > wrap.workers:
        # Limit ourselves to one process a second (over the period of
        # number of workers * 1 second). This will allow workers to
        # start up quickly but ensure we don't fork off children that
        # die instantly too quickly.
        if time.time() - wrap.forktimes[0] < wrap.workers:
            LOG.info('Forking too fast, sleeping')
            time.sleep(1)

        wrap.forktimes.pop(0)

    wrap.forktimes.append(time.time())

    pid = os.fork()
    if pid == 0:
        self.launcher = self._child_process(wrap.service)
        while True:
            self._child_process_handle_signal()
            status, signo = self._child_wait_for_exit_or_signal(
                self.launcher)
            if not _is_sighup_and_daemon(signo):
                self.launcher.wait()
                break
            self.launcher.restart()

        os._exit(status)

    LOG.debug('Started child %d', pid)

    wrap.children.add(pid)
    self.children[pid] = wrap

看見熟悉的fork沒有靖榕，只是簡(jiǎn)單的調(diào)用了一個(gè)os.fork()标锄，然后子進(jìn)程開始運(yùn)行，子進(jìn)程調(diào)用_child_process

def _child_process(self, service):
    self._child_process_handle_signal()

    # Reopen the eventlet hub to make sure we don't share an epoll
    # fd with parent and/or siblings, which would be bad
    eventlet.hubs.use_hub()

    # Close write to ensure only parent has it open
    os.close(self.writepipe)
    # Create greenthread to watch for parent to close pipe
    eventlet.spawn_n(self._pipe_watcher)

    # Reseed random number generator
    random.seed()

    launcher = Launcher(self.conf, restart_method=self.restart_method)
    launcher.launch_service(service)
    return launcher

_child_process其實(shí)很簡(jiǎn)單茁计，創(chuàng)建一個(gè)Launcher料皇，調(diào)用Laucher.launch_service方法谓松，前面介紹過，其實(shí)ServiceLauncher繼承自Launcher践剂，也是調(diào)用的launcher_service方法鬼譬，將服務(wù)啟動(dòng)，因此接下來的步驟可以參考前面逊脯，最終都將調(diào)用service.start方法啟動(dòng)服務(wù)

三优质、WSGIService的啟動(dòng)—續(xù)

回到前面的啟動(dòng)部分，從launcher節(jié)的說明军洼，我們知道服務(wù)的啟動(dòng)最終調(diào)用了service的start方法巩螃，而這里的service就是我們最開始在api.py中創(chuàng)建的service，然后一層層傳進(jìn)后面的啟動(dòng)器中的匕争，我們繼續(xù)回到WSGIService類中的start(self)方法

def start(self):
    …
    if self.manager:
        self.manager.init_host()
    self.server.start()
    self.port = self.server.port

這里調(diào)用了oslo_service/wsgi.py中的start(self)方法

def start(self):
    …
    self.dup_socket = self.socket.dup()

    if self._use_ssl:
        self.dup_socket = sslutils.wrap(self.conf, self.dup_socket)

    wsgi_kwargs = {
        'func': eventlet.wsgi.server,
        'sock': self.dup_socket,
        'site': self.app,
        'protocol': self._protocol,
        'custom_pool': self._pool,
        'log': self._logger,
        'log_format': self.conf.wsgi_log_format,
        'debug': False,
        'keepalive': self.conf.wsgi_keep_alive,
        'socket_timeout': self.client_socket_timeout
        }

    if self._max_url_len:
        wsgi_kwargs['url_length_limit'] = self._max_url_len

    self._server = eventlet.spawn(**wsgi_kwargs)

注意 wsgi_kwargs 中的參數(shù) func避乏，它的值為 eventlet.wsgi.server，在?eventlet/wsgi.py?的定義如下：

def server(sock, site,
    …
     try:
        serv.log.info("(%s) wsgi starting up on %s" % (
            serv.pid, socket_repr(sock)))
        while is_accepting:
            try:
                client_socket = sock.accept()
                client_socket[0].settimeout(serv.socket_timeout)
                serv.log.debug("(%s) accepted %r" % (
                    serv.pid, client_socket[1]))
                try:
                    pool.spawn_n(serv.process_request, client_socket)
                except AttributeError:
                    warnings.warn("wsgi's pool should be an instance of "
                                  "eventlet.greenpool.GreenPool, is %s. Please convert your"
                                  " call site to use GreenPool instead" % type(pool),
                                  DeprecationWarning, stacklevel=2)
                    pool.execute_async(serv.process_request, client_socket)
            except ACCEPT_EXCEPTIONS as e:
                if support.get_errno(e) not in ACCEPT_ERRNO:
                    raise
            except (KeyboardInterrupt, SystemExit):
                serv.log.info("wsgi exiting")
                break
    finally:
        pool.waitall()        …

看甘桑，是不是看到熟悉的一幕了拍皮！sock.accept() 監(jiān)聽請(qǐng)求，每當(dāng)接收到一個(gè)新請(qǐng)求跑杭，調(diào)用 pool.spawn_n() 啟動(dòng)一個(gè)協(xié)程處理該請(qǐng)求

四铆帽、Service的啟動(dòng)

Service 類型的進(jìn)程同樣由 nova/cmd/* 目錄下某些文件創(chuàng)建：

• nova-schedule: nova/cmd/schedule.py

• ……
  作為消息中間件的消費(fèi)者，它們監(jiān)聽各自的 queue艘蹋，每當(dāng)有 rpc 請(qǐng)求來臨時(shí)锄贼，它們創(chuàng)建一個(gè)新的協(xié)程處理 rpc 請(qǐng)求。以nova-schedule為例女阀，啟動(dòng)時(shí)初始化一個(gè) Server(由?service.py?定義) 對(duì)象宅荤。整個(gè)Launcher過程跟WSGIServer一樣，只是service的start()有些區(qū)別而已

def start(self):
    …
    target = messaging.Target(topic=self.topic, server=self.host)
    endpoints = [self.manager]
    endpoints.extend(self.manager.additional_endpoints)
    serializer = objects_base.KarborObjectSerializer()
    self.rpcserver = rpc.get_server(target, endpoints, serializer)
    self.rpcserver.start()

經(jīng)過層層調(diào)用浸策，最終生成了這樣一個(gè)RPCServer對(duì)象

class RPCServer(msg_server.MessageHandlingServer):
    def __init__(self, transport, target, dispatcher, executor='blocking'):
        super(RPCServer, self).__init__(transport, dispatcher, executor)
        self._target = target

該類繼承自MessageHandlingServer冯键；
注：nova 的各個(gè)組件都依賴 oslo.messaging 訪問消息服務(wù)器，通過?oslo/messaging/server.py?初始化一個(gè) MessageHandlingServer 的對(duì)象庸汗，監(jiān)聽消息隊(duì)列惫确。最終調(diào)用了該service的start方法

def start(self, override_pool_size=None):
    …
    if self._started:
        LOG.warning(_LW('Restarting a MessageHandlingServer is inherently '
                        'racy. It is deprecated, and will become a noop '
                        'in a future release of oslo.messaging. If you '
                        'need to restart MessageHandlingServer you should '
                        'instantiate a new object.'))
    self._started = True

    executor_opts = {}

    if self.executor_type in ("threading", "eventlet"):
        executor_opts["max_workers"] = (
            override_pool_size or self.conf.executor_thread_pool_size
        )
    self._work_executor = self._executor_cls(**executor_opts)

    try:
        self.listener = self._create_listener()
    except driver_base.TransportDriverError as ex:
        raise ServerListenError(self.target, ex)

    # HACK(sileht): We temporary pass the executor to the rabbit
    # listener to fix a race with the deprecated blocking executor.
    # We do this hack because this is need only for 'synchronous'
    # executor like blocking. And this one is deprecated. Making
    # driver working in an sync and an async way is complicated
    # and blocking have 0% tests coverage.
    if hasattr(self.listener, '_poll_style_listener'):
        l = self.listener._poll_style_listener
        if hasattr(l, "_message_operations_handler"):
            l._message_operations_handler._executor = (
                self.executor_type)

    self.listener.start(self._on_incoming)

上述的對(duì)象又初始化一個(gè) EventletExecutor(由?oslo/messaging/_executors/impl_eventlet.py) 類型的 excuete 對(duì)象，它調(diào)用?self.listener.poll()?監(jiān)聽 rpc 請(qǐng)求蚯舱，每當(dāng)接收到一個(gè)請(qǐng)求改化，創(chuàng)建一個(gè)協(xié)程處理該請(qǐng)求。

class EventletExecutor(base.ExecutorBase):
    ......

    def start(self):
        if self._thread is not None:
            return

        @excutils.forever_retry_uncaught_exceptions
        def _executor_thread():
            try:
                while True:
                    incoming = self.listener.poll()
                    spawn_with(ctxt=self.dispatcher(incoming),
                               pool=self._greenpool)
            except greenlet.GreenletExit:
                return

        self._thread = eventlet.spawn(_executor_thread)

博客：https://tunsuy.github.io/

github：https://github.com/tunsuy