OpenStack-RPC-server的构建(二)

1. server= service.Service.create(…)中的create方法是/usr/lib/python2.7/site-packages/nova/service.py文件中Service类的类方法，用于创建且返回一个Service对象。

#/usr/lib/python2.7/site-packages/nova/service.py
@classmethod
def create(cls, host=None, binary=None, topic=None, manager=None,
report_interval=None, periodic_enable=None,
periodic_fuzzy_delay=None, periodic_interval_max=None,
db_allowed=True):
"""Instantiates class and passes back application object.

:param host: defaults to CONF.host
:param binary: defaults to basename of executable
:param topic: defaults to bin_name - 'nova-' part
:param manager: defaults to CONF.<topic>_manager
:param report_interval: defaults to CONF.report_interval
:param periodic_enable: defaults to CONF.periodic_enable
:param periodic_fuzzy_delay: defaults to CONF.periodic_fuzzy_delay
:param periodic_interval_max: if set, the max time to wait between runs

"""
if not host:
host = CONF.host
if not binary:
binary = os.path.basename(sys.argv[0])
if not topic:
topic = binary.rpartition('nova-')[2]
if not manager:
manager_cls = ('%s_manager' %
binary.rpartition('nova-')[2])
manager = CONF.get(manager_cls, None)
if report_interval is None:
report_interval = CONF.report_interval
if periodic_enable is None:
periodic_enable = CONF.periodic_enable
if periodic_fuzzy_delay is None:
periodic_fuzzy_delay = CONF.periodic_fuzzy_delay

debugger.init()

service_obj = cls(host, binary, topic, manager,
report_interval=report_interval,
periodic_enable=periodic_enable,
periodic_fuzzy_delay=periodic_fuzzy_delay,
periodic_interval_max=periodic_interval_max,
db_allowed=db_allowed)

return service_obj

其中，host为系统的hostname(在我的机器上为’jun’)，binary为’nova-scheduler’，topic为CONF.scheduler_topic(即scheduler)，这里需要说明的是：

# /usr/lib/python2.7/site-packages/nova/scheduler/rpcapi.py
rpcapi_opts = [
cfg.StrOpt('scheduler_topic',
default='scheduler',
help='The topic scheduler nodes listen on'),
]……………………………………………………………………………………………………………………………………………………………………………………….(1)
#/usr/lib/python2.7/site-packages/nova/cmd/scheduler.py
CONF = cfg.CONF
CONF.import_opt('scheduler_topic', 'nova.scheduler.rpcapi')…………………………….(2)
#/usr/lib/python2.7/site-packages/nova/cmd/scheduler.py
server = service.Service.create(binary='nova-scheduler',
topic=CONF.scheduler_topic)………………………………(3)

在(2)中通过import_opt来申明在nova.scheduler.rpcapi模块中定义的配置选项，

所以(3)中的CONF.scheduler_topic即为(1)中的default值：’scheduler’

manager的值为：nova.scheduler.manager.SchedulerManager，这里只是一个名词而已，还并未真正构造SchedulerManager对象。

service_obj = cls(host, binary, topic,…)开始构造Service对象。类Service的构造函数如下所示：

#/usr/lib/python2.7/site-packages/nova/service.py
class Service(service.Service):
"""Service object for binaries running on hosts.

A service takes a manager and enables rpc by listening to queues based
on topic. It also periodically runs tasks on the manager and reports
it state to the database services table.
"""

def __init__(self, host, binary, topic, manager, report_interval=None,
periodic_enable=None, periodic_fuzzy_delay=None,
periodic_interval_max=None, db_allowed=True,
*args, **kwargs):
super(Service, self).__init__()
self.host = host
self.binary = binary
self.topic = topic
self.manager_class_name = manager
# NOTE(russellb) We want to make sure to create the servicegroup API
# instance early, before creating other things such as the manager,
# that will also create a servicegroup API instance.  Internally, the
# servicegroup only allocates a single instance of the driver API and
# we want to make sure that our value of db_allowed is there when it
# gets created.  For that to happen, this has to be the first instance
# of the servicegroup API.
self.servicegroup_api = servicegroup.API(db_allowed=db_allowed)
manager_class = importutils.import_class(self.manager_class_name)
#构造nova.scheduler.manager.SchedulerManager对象
self.manager = manager_class(host=self.host, *args, **kwargs)
self.rpcserver = None
self.report_interval = report_interval
self.periodic_enable = periodic_enable
self.periodic_fuzzy_delay = periodic_fuzzy_delay
self.periodic_interval_max = periodic_interval_max
self.saved_args, self.saved_kwargs = args, kwargs
self.backdoor_port = None
#因为db_allowed值为True，所以构造nova.conductor.api.LocalAPI对象，
#该对象能直接访问数据库，不需通过RPC的方式，所以Nova-scheduler
#组件能够直接访问数据库。
self.conductor_api = conductor.API(use_local=db_allowed)
self.conductor_api.wait_until_ready(context.get_admin_context())

2. service.serve(server)

#/usr/lib/python2.7/site-packages/nova/service.py
def serve(server, workers=None):
global _launcher
if _launcher:
raise RuntimeError(_('serve() can only be called once'))

_launcher = service.launch(server, workers=workers)
#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py
def launch(service, workers=1):
if workers is None or workers == 1:
launcher = ServiceLauncher()
launcher.launch_service(service)
else:
launcher = ProcessLauncher()
launcher.launch_service(service, workers=workers)

return launcher

由于works=None，所以在launch方法中执行红色部分，即launcher为ServiceLauncher对象，然后在launch_service方法执行一些操作后返回该对象。那么launch_service方法对了哪些操作呢？我们继续向下分析。

#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py
def launch_service(self, service):
"""Load and start the given service.

:param service: The service you would like to start.
:returns: None

"""
service.backdoor_port = self.backdoor_port
#将传进来的service对象加载到Services对象中(self.services = Services())
self.services.add(service)
#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py
class Services(object):

def __init__(self):
self.services = []
#创建ThreadGroup对象，其中该对象有绿色线程池属性(即如下的self.pool)
self.tg = threadgroup.ThreadGroup()
self.done = event.Event()

def add(self, service):
self.services.append(service)
#将service运行时要执行的方法(self.run_service)加载到绿色线程中，其中
#当执行run_service方法时，service和self.done作为run_service的参数。
self.tg.add_thread(self.run_service, service, self.done)

#/usr/lib/python2.7/site-packages/nova/openstack/common/threadgroup.py
class ThreadGroup(object):
"""The point of the ThreadGroup class is to:

* keep track of timers and greenthreads (making it easier to stop them
when need be).
* provide an easy API to add timers.
"""
def __init__(self, thread_pool_size=10):
self.pool = greenpool.GreenPool(thread_pool_size)
self.threads = []
self.timers = []
def add_thread(self, callback, *args, **kwargs):
#这里的callback就是上面的self.run_service，当线程启动时，将会执行
#self.run_service方法。
gt = self.pool.spawn(callback, *args, **kwargs)
th = Thread(gt, self)
self.threads.append(th)
return th

根据上面的分析，launch_service方法会为传进来的每个service分配一个绿色线程，且绿色线程启动时，将会去执行self.run_service方法，且将service和self.done作为参数传给run_service方法去执行。目前，新建的绿色线程只是标记为可调度，并不会被立即调度执行，只有当主线程执行到gt.wait()时，这个绿色线程才有机会被调度去执行run_service函数。
3 service.wait()

#/usr/lib/python2.7/site-packages/nova/ service.py
def wait():
_launcher.wait()

#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py
def wait(self, ready_callback=None):
systemd.notify_once()
while True:
self.handle_signal()
status, signo = self._wait_for_exit_or_signal(ready_callback)
if not _is_sighup_and_daemon(signo):
return status
self.restart()
def _wait_for_exit_or_signal(self, ready_callback=None):
status = None
signo = 0

LOG.debug('Full set of CONF:')
CONF.log_opt_values(LOG, logging.DEBUG)

try:
if ready_callback:
ready_callback()
super(ServiceLauncher, self).wait()
except SignalExit as exc:
signame = _signo_to_signame(exc.signo)
LOG.info(_LI('Caught %s, exiting'), signame)
status = exc.code
signo = exc.signo
except SystemExit as exc:
status = exc.code
finally:
self.stop()

return status, signo

这里我们从_wait_for_exit_or_signal方法中的super(ServiceLauncher, self).wait()往下看(这里是调用父类的wait方法)

#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py:Launcher
def wait(self):
"""Waits until all services have been stopped, and then returns.

:returns: None

"""
self.services.wait()

#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py:Services
def wait(self):
self.tg.wait()

#/usr/lib/python2.7/site-packages/nova/openstack/common/threadgroup.py:ThreadGroup
def wait(self):
for x in self.timers:
try:
x.wait()
except eventlet.greenlet.GreenletExit:
pass
except Exception as ex:
LOG.exception(ex)
current = threading.current_thread()

# Iterate over a copy of self.threads so thread_done doesn't
# modify the list while we're iterating
for x in self.threads[:]:
if x is current:
continue
try:
x.wait()
except eventlet.greenlet.GreenletExit:
pass
except Exception as ex:
LOG.exception(ex)

#/usr/lib/python2.7/site-packages/nova/openstack/common/threadgroup.py:Thread
def wait(self):
return self.thread.wait()

此时，我们已经调用到最底层了，此后该绿色线程被调度去执行run_service函数。

#/usr/lib/python2.7/site-packages/nova/openstack/common/service.py
@staticmethod
def run_service(service, done):
"""Service start wrapper.

:param service: service to run
:param done: event to wait on until a shutdown is triggered
:returns: None

"""
service.start()
done.wait()

#/usr/lib/python2.7/site-packages/nova/service.py
def start(self):
verstr = version.version_string_with_package()
LOG.info(_LI('Starting %(topic)s node (version %(version)s)'),
{'topic': self.topic, 'version': verstr})
self.basic_config_check()
self.manager.init_host()
self.model_disconnected = False
ctxt = context.get_admin_context()
try:
self.service_ref = (
self.conductor_api.service_get_by_host_and_binary(
ctxt, self.host, self.binary))
self.service_id = self.service_ref['id']
except exception.NotFound:
try:
self.service_ref = self._create_service_ref(ctxt)
except (exception.ServiceTopicExists,
exception.ServiceBinaryExists):
# NOTE(danms): If we race to create a record with a sibling
# worker, don't fail here.
self.service_ref = (
self.conductor_api.service_get_by_host_and_binary(
ctxt, self.host, self.binary))

self.manager.pre_start_hook()

if self.backdoor_port is not None:
self.manager.backdoor_port = self.backdoor_port

LOG.debug("Creating RPC server for service %s", self.topic)

target = messaging.Target(topic=self.topic, server=self.host)

endpoints = [
self.manager,
baserpc.BaseRPCAPI(self.manager.service_name, self.backdoor_port)
]
endpoints.extend(self.manager.additional_endpoints)

serializer = objects_base.NovaObjectSerializer()

self.rpcserver = rpc.get_server(target, endpoints, serializer)
self.rpcserver.start()

self.manager.post_start_hook()

LOG.debug("Join ServiceGroup membership for this service %s",
self.topic)
# Add service to the ServiceGroup membership group.
self.servicegroup_api.join(self.host, self.topic, self)

if self.periodic_enable:
if self.periodic_fuzzy_delay:
initial_delay = random.randint(0, self.periodic_fuzzy_delay)
else:
initial_delay = None

self.tg.add_dynamic_timer(self.periodic_tasks,
initial_delay=initial_delay,
periodic_interval_max=
self.periodic_interval_max)

self.conductor_api.service_get_by_host_and_binary方法为根据host和binary值查询数据库中的信息，由于host为’jun’，且binary的值为’nova-scheduler’,所以它将查询下面数据库中的nova-scheduler信息，并将其转换成字典形式返回给变量self.service_ref。

MariaDB [nova]> select * from services;
+---------------------+---------------------+------------+----+------+------------------+-------------+--------------+----------+---------+-----------------+
| created_at          | updated_at          | deleted_at | id | host | binary           | topic       | report_count | disabled | deleted | disabled_reason |
+---------------------+---------------------+------------+----+------+------------------+-------------+--------------+----------+---------+-----------------+
| 2015-08-28 14:06:37 | 2015-12-17 13:39:14 | NULL       |  1 | jun  | nova-consoleauth | consoleauth |        80765 |        0 |       0 | NULL            |
| 2015-08-28 14:06:41 | 2015-12-17 13:39:17 | NULL       |  2 | jun  | nova-scheduler   | scheduler   |        80436 |        0 |       0 | NULL            |
| 2015-08-28 14:07:05 | 2015-12-17 13:39:17 | NULL       |  3 | jun  | nova-conductor   | conductor   |        80751 |        0 |       0 | NULL            |
| 2015-08-28 14:07:18 | 2015-12-13 14:50:42 | NULL       |  7 | jun2 | nova-compute     | compute     |        10078 |        0 |       0 | NULL            |
| 2015-08-28 14:08:40 | 2015-12-17 13:39:14 | NULL       |  8 | jun  | nova-cert        | cert        |        80797 |        0 |       0 | NULL            |
+---------------------+---------------------+------------+----+------+------------------+-------------+--------------+----------+---------+-----------------+
5 rows in set (0.00 sec)

如果数据库中没有该信息或没有查询到该信息，则它会捕获到一个exception.NotFound的异常，此时它会根据host和binary信息重新创建一条与与上面红色数据库相似的信息，同时，在创建数据库信息的时候，它会检测上述的数据库信息是否已经存在，如果存在(此时抛出exception.ServiceTopicExists或exception.ServiceBinaryExists)，它就不会创建(因为当时抛出exception.NotFound异常的时候或许是因为没有查询到该数据库信息，并不能说明该数据库信息不存在数据库中。所以在重新创建数据库信息的时候，它会检测要添加的数据库信息是否已经存在数据库中)，然后再次查询数据库信息。如果数据库中的确没有该信息，这样数据库才会更新信息，且将更新的信息以字典的形式返回给变量self.service_ref.

target =messaging.Target(topic=self.topic, server=self.host)这里是创建RPC-server的一个Target对象，根据我们《OpenStack的oslo_messaging组件使用》文章中所描述的：创建RPC-server时，需指明topic和server参数，exchange可选。

endpoints从目前的代码来看，共添加了3个对象到endpoints中，两个基本的endpoints(self.manager和baserpc.BaseRPCAPI对象)，一个扩展的endpoints(self.manager.additional_endpoints)，其中目前扩展的endpoints中只有一个_SchedulerManagerV3Proxy对象在additional_endpoints列表中，因为self.manager是nova.scheduler.manager.SchedulerManager对象，所以查看创建SchedulerManager对象的初始化代码，代码如下：

#/usr/lib/python2.7/site-packages/nova/scheduler/manager.py
class SchedulerManager(manager.Manager):
"""Chooses a host to run instances on."""

target = messaging.Target(version='4.2')

def __init__(self, scheduler_driver=None, *args, **kwargs):
if not scheduler_driver:
scheduler_driver = CONF.scheduler_driver
self.driver = importutils.import_object(scheduler_driver)
super(SchedulerManager, self).__init__(service_name='scheduler',
*args, **kwargs)
self.additional_endpoints.append(_SchedulerManagerV3Proxy(self))

根据红色标记部分可知，self.additional_endpoints是只有_SchedulerManagerV3Proxy对象的列表。

因此，在这3个endpoints中的所有方法都能被连接到RPC-server上的RPC-client通过Transport对象远程调用。

serializer是用于序列化/反序列化消息的。

继续回到/usr/lib/python2.7/site-packages/nova/service.py的start方法中，其中self.rpcserver = rpc.get_server(target,endpoints, serializer)才真正的创建RPC-server。

#/usr/lib/python2.7/site-packages/nova/rpc.py
def get_server(target, endpoints, serializer=None):
#TRANSPORT对象在执行rpc.init(CONF)时就已经创建成功
assert TRANSPORT is not None
serializer = RequestContextSerializer(serializer)
return messaging.get_rpc_server(TRANSPORT,
target,
endpoints,
executor='eventlet',
serializer=serializer)

#/usr/lib/python2.7/site-packages/oslo_messaging/rpc/server.py
def get_rpc_server(transport, target, endpoints,
executor='blocking', serializer=None):
"""Construct an RPC server.

The executor parameter controls how incoming messages will be received and
dispatched. By default, the most simple executor is used - the blocking
executor.

If the eventlet executor is used, the threading and time library need to be
monkeypatched.

:param transport: the messaging transport
:type transport: Transport
:param target: the exchange, topic and server to listen on
:type target: Target
:param endpoints: a list of endpoint objects
:type endpoints: list
:param executor: name of a message executor - for example
'eventlet', 'blocking'
:type executor: str
:param serializer: an optional entity serializer
:type serializer: Serializer
"""
dispatcher = rpc_dispatcher.RPCDispatcher(target, endpoints, serializer)
return msg_server.MessageHandlingServer(transport, dispatcher, executor)

其中在get_rpc_server方法中，构造一个RPCDispatcher对象，该对象以参数形式用于构造MessageHandlingServer对象，在RPC-server监听到有消息到来时，这个dispatcher将被调用用于分发message。

总结：本篇文章我们分析重点分析了server =service.Service.create(binary='nova-scheduler',topic=CONF.scheduler_topic)和service.serve(server)的调用，对于service.wait()方法中的start方法需要涉及包括oslo_messaging，kombu和amqp层面的函数调用，分析的代码较多，我们从下一篇文章具体分析。