cinder服务启动源码分析章

本文以Newton版本的cinder为例

一、cinder api服务

根据setup.cfg找到cinder-api服务的启动入口

#setup.cfg
[entry_points]
console_scripts =
    cinder-api = cinder.cmd.api:main

#cinder/cmd/api.py
def main():
    # 加载辅助对象，封装与数据库相关的操作
    objects.register_all()
    gmr_opts.set_defaults(CONF)
    # 加载配置并设置日志
    CONF(sys.argv[1:], project='cinder',
         version=version.version_string())
    config.set_middleware_defaults()
    logging.setup(CONF, "cinder")
    python_logging.captureWarnings(True)
    utils.monkey_patch()

    gmr.TextGuruMeditation.setup_autorun(version, conf=CONF)
    # 初始化rpc
    rpc.init(CONF)
    launcher = service.process_launcher()
    
    # 创建一个WSGIService对象
    server = service.WSGIService('osapi_volume')
    launcher.launch_service(server, workers=server.workers)
    launcher.wait()

WSGIService关键代码如下：

#cinder/service.py
class WSGIService(service.ServiceBase):
    def _get_manager(self):
        
        fl = '%s_manager' % self.name
        if fl not in CONF:
            return None

        manager_class_name = CONF.get(fl, None)
        if not manager_class_name:
            return None

        manager_class = importutils.import_class(manager_class_name)
        return manager_class()
    def __init__(self, name, loader=None):
       
        ...
        self.name = name
        # 会加载名为`osapi_volume_manager`的管理器（或None）
        self.manager = self._get_manager()
        """创建WSGI应用加载器
        并根据配置文件(`cinder.conf`)设置应用配置路径:
        `config_path` = `/etc/cinder/paste-api.ini`
        """
        self.loader = loader or wsgi.Loader(CONF)
        """ 加载‘/etc/cinder/paste-api.ini’文件中‘[composite:osapi_volume]’中定义的app,
        具体的代码看下面
        self.app = self.loader.load_app(name)
        """根据配置文件(`cinder.conf`)设置监听地址及工作线程数
        如果未指定监听ip及端口就分别设置为`0.0.0.0`及`0`
        如果为指定工作线程数就设置为cpu个数
        如果设置的工作线程数小于1，则抛异常
        """
        self.host = getattr(CONF, '%s_listen' % name, "0.0.0.0")
        self.port = getattr(CONF, '%s_listen_port' % name, 0)
        self.use_ssl = getattr(CONF, '%s_use_ssl' % name, False)
        self.workers = (getattr(CONF, '%s_workers' % name, None) or
                        processutils.get_worker_count())
        if self.workers and self.workers < 1:
            worker_name = '%s_workers' % name
            msg = (_("%(worker_name)s value of %(workers)d is invalid, "
                     "must be greater than 0.") %
                   {'worker_name': worker_name,
                    'workers': self.workers})
            raise exception.InvalidInput(msg)
            
        """如果CONF.profiler.profiler_enabled = True就开启性能分析  
        创建一个类型为`Messaging`的通知器(`_notifier`)，将性能数据发送给
        ceilometer
        """
        setup_profiler(name, self.host)
          
        # 创建WSGI Server对象
        self.server = wsgi.Server(CONF,
                                  name,
                                  self.app,
                                  host=self.host,
                                  port=self.port,
                                  use_ssl=self.use_ssl)

load_app的代码如下所示：

#oslo.service/wsgi.py
class Loader(object):
    """Used to load WSGI applications from paste configurations."""

    def __init__(self, conf):
        """Initialize the loader, and attempt to find the config.
        :param conf: Application config
        :returns: None
        """
        conf.register_opts(_options.wsgi_opts)
        self.config_path = None

        #根据配置文件获得config_path的路径
        config_path = conf.api_paste_config
        if not os.path.isabs(config_path):
            self.config_path = conf.find_file(config_path)
        elif os.path.exists(config_path):
            self.config_path = config_path

        if not self.config_path:
            raise ConfigNotFound(path=config_path)

    def load_app(self, name):
        """Return the paste URLMap wrapped WSGI application.
        :param name: Name of the application to load.
        :returns: Paste URLMap object wrapping the requested application.
        :raises: PasteAppNotFound
        """
        try:
            LOG.debug("Loading app %(name)s from %(path)s",
                      {'name': name, 'path': self.config_path})
            return deploy.loadapp("config:%s" % self.config_path, name=name)
        except LookupError:
            LOG.exception("Couldn't lookup app: %s", name)
            raise PasteAppNotFound(name=name, path=self.config_path)

load_app完成了如下操作:
1.根据/etc/cinder/api-paste.ini中的Paste规则加载相应的过滤器和app
2.设置路由映射(Router)

这些具体的过程可以参考我的另一篇博客： 深入理解nova api服务

上面的内容是创建创建一个WSGIService对象并初始化的过程，接着往下分析：

#cinder/cmd/api.py
def main():
    ...
    #通过进程启动器 - 在新的进程中启动服务程序，并等待服务启动结束
    #最终会调用`WSGIService.start`方法启动服务
    launcher.launch_service(server, workers=server.workers)
    
#cinder/service.py
from oslo_service import wsgi

class WSGIService(service.ServiceBase):
    ...
    def start(self):
        ...
        if self.manager:
            self.manager.init_host()
        """这里的server就是上面初始化时创建的wsgi Server对象
        self.server = wsgi.Server(CONF,
                                  name,
                                  self.app,
                                  host=self.host,
                                  port=self.port,
                                  use_ssl=self.use_ssl)
        """
        self.server.start()
        self.port = self.server.port
        

#接上文的start方法
#oslo.service/wsgi.py
class Server(service.ServiceBase):
    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)

这样，cinder-api中的socket监听就准备就绪了：基于配置cinder-api会启动多个监听线程，每个客户端连接用来一个线程来处理.

二、cinder scheduler服务

同理，找到启动入口：

#cinder/cmd/scheduler.py
def main():
    objects.register_all()
    gmr_opts.set_defaults(CONF)
    CONF(sys.argv[1:], project='cinder',
         version=version.version_string())
    logging.setup(CONF, "cinder")
    python_logging.captureWarnings(True)
    utils.monkey_patch()
    gmr.TextGuruMeditation.setup_autorun(version, conf=CONF)
    server = service.Service.create(binary='cinder-scheduler')
    service.serve(server)
    service.wait()

service.Service.create部分代码如下：

class Service(service.Service):
    @classmethod
    def create(cls, host=None, binary=None, topic=None, manager=None,
               report_interval=None, periodic_interval=None,
               periodic_fuzzy_delay=None, service_name=None,
               coordination=False, cluster=None):
        ...
        if not host:
            host = CONF.host
        if not binary:
            binary = os.path.basename(inspect.stack()[-1][1])
        if not topic:
            topic = binary
        if not manager:
            subtopic = topic.rpartition('cinder-')[2]
            manager = CONF.get('%s_manager' % subtopic, None)
        #实例化Service类,会执行Service类的__init__方法
        service_obj = cls(host, binary, topic, manager,
                          report_interval=report_interval,
                          periodic_interval=periodic_interval,
                          periodic_fuzzy_delay=periodic_fuzzy_delay,
                          service_name=service_name,
                          coordination=coordination,
                          cluster=cluster)

        return service_obj

接着是service.serve(server)的部分代码：

#cinder/cmd/scheduler.py
service.serve(server)


#cinder/service.py
def serve(server, workers=None):
    global _launcher
    if _launcher:
        raise RuntimeError(_('serve() can only be called once'))

    _launcher = service.launch(CONF, server, workers=workers)
    
#最终会调用Service类的start方法,部分代码如下
class Service(service.Service):

    def start(self):
    
        ctxt = context.get_admin_context()
        endpoints = [self.manager]
        endpoints.extend(self.manager.additional_endpoints)
        obj_version_cap = objects.Service.get_minimum_obj_version(ctxt)
        

        target = messaging.Target(topic=self.topic, server=self.host)
        self.rpcserver = rpc.get_server(target, endpoints, serializer)
        #创建RPC连接，启动消费者线程,等待队列消息
        self.rpcserver.start()